Adding upstream version 15.4.upstream/15.4upstream

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

1 files changed, 713 insertions, 0 deletions

diff --git a/src/bin/psql/crosstabview.c b/src/bin/psql/crosstabview.c
new file mode 100644
index 0000000..2c29138
--- /dev/null
+++ b/src/bin/psql/crosstabview.c
@@ -0,0 +1,713 @@
+/*
+ * psql - the PostgreSQL interactive terminal
+ *
+ * Copyright (c) 2000-2022, PostgreSQL Global Development Group
+ *
+ * src/bin/psql/crosstabview.c
+ */
+#include "postgres_fe.h"
+
+#include "common.h"
+#include "common/logging.h"
+#include "crosstabview.h"
+#include "pqexpbuffer.h"
+#include "psqlscanslash.h"
+#include "settings.h"
+
+/*
+ * Value/position from the resultset that goes into the horizontal or vertical
+ * crosstabview header.
+ */
+typedef struct _pivot_field
+{
+	/*
+	 * Pointer obtained from PQgetvalue() for colV or colH. Each distinct
+	 * value becomes an entry in the vertical header (colV), or horizontal
+	 * header (colH). A Null value is represented by a NULL pointer.
+	 */
+	char	   *name;
+
+	/*
+	 * When a sort is requested on an alternative column, this holds
+	 * PQgetvalue() for the sort column corresponding to <name>. If <name>
+	 * appear multiple times, it's the first value in the order of the results
+	 * that is kept. A Null value is represented by a NULL pointer.
+	 */
+	char	   *sort_value;
+
+	/*
+	 * Rank of this value, starting at 0. Initially, it's the relative
+	 * position of the first appearance of <name> in the resultset. For
+	 * example, if successive rows contain B,A,C,A,D then it's B:0,A:1,C:2,D:3
+	 * When a sort column is specified, ranks get updated in a final pass to
+	 * reflect the desired order.
+	 */
+	int			rank;
+} pivot_field;
+
+/* Node in avl_tree */
+typedef struct _avl_node
+{
+	/* Node contents */
+	pivot_field field;
+
+	/*
+	 * Height of this node in the tree (number of nodes on the longest path to
+	 * a leaf).
+	 */
+	int			height;
+
+	/*
+	 * Child nodes. [0] points to left subtree, [1] to right subtree. Never
+	 * NULL, points to the empty node avl_tree.end when no left or right
+	 * value.
+	 */
+	struct _avl_node *children[2];
+} avl_node;
+
+/*
+ * Control structure for the AVL tree (binary search tree kept
+ * balanced with the AVL algorithm)
+ */
+typedef struct _avl_tree
+{
+	int			count;			/* Total number of nodes */
+	avl_node   *root;			/* root of the tree */
+	avl_node   *end;			/* Immutable dereferenceable empty tree */
+} avl_tree;
+
+
+static bool printCrosstab(const PGresult *result,
+						  int num_columns, pivot_field *piv_columns, int field_for_columns,
+						  int num_rows, pivot_field *piv_rows, int field_for_rows,
+						  int field_for_data);
+static void avlInit(avl_tree *tree);
+static void avlMergeValue(avl_tree *tree, char *name, char *sort_value);
+static int	avlCollectFields(avl_tree *tree, avl_node *node,
+							 pivot_field *fields, int idx);
+static void avlFree(avl_tree *tree, avl_node *node);
+static void rankSort(int num_columns, pivot_field *piv_columns);
+static int	indexOfColumn(char *arg, const PGresult *res);
+static int	pivotFieldCompare(const void *a, const void *b);
+static int	rankCompare(const void *a, const void *b);
+
+
+/*
+ * Main entry point to this module.
+ *
+ * Process the data from *res according to the options in pset (global),
+ * to generate the horizontal and vertical headers contents,
+ * then call printCrosstab() for the actual output.
+ */
+bool
+PrintResultInCrosstab(const PGresult *res)
+{
+	bool		retval = false;
+	avl_tree	piv_columns;
+	avl_tree	piv_rows;
+	pivot_field *array_columns = NULL;
+	pivot_field *array_rows = NULL;
+	int			num_columns = 0;
+	int			num_rows = 0;
+	int			field_for_rows;
+	int			field_for_columns;
+	int			field_for_data;
+	int			sort_field_for_columns;
+	int			rn;
+
+	avlInit(&piv_rows);
+	avlInit(&piv_columns);
+
+	if (PQresultStatus(res) != PGRES_TUPLES_OK)
+	{
+		pg_log_error("\\crosstabview: statement did not return a result set");
+		goto error_return;
+	}
+
+	if (PQnfields(res) < 3)
+	{
+		pg_log_error("\\crosstabview: query must return at least three columns");
+		goto error_return;
+	}
+
+	/* Process first optional arg (vertical header column) */
+	if (pset.ctv_args[0] == NULL)
+		field_for_rows = 0;
+	else
+	{
+		field_for_rows = indexOfColumn(pset.ctv_args[0], res);
+		if (field_for_rows < 0)
+			goto error_return;
+	}
+
+	/* Process second optional arg (horizontal header column) */
+	if (pset.ctv_args[1] == NULL)
+		field_for_columns = 1;
+	else
+	{
+		field_for_columns = indexOfColumn(pset.ctv_args[1], res);
+		if (field_for_columns < 0)
+			goto error_return;
+	}
+
+	/* Insist that header columns be distinct */
+	if (field_for_columns == field_for_rows)
+	{
+		pg_log_error("\\crosstabview: vertical and horizontal headers must be different columns");
+		goto error_return;
+	}
+
+	/* Process third optional arg (data column) */
+	if (pset.ctv_args[2] == NULL)
+	{
+		int			i;
+
+		/*
+		 * If the data column was not specified, we search for the one not
+		 * used as either vertical or horizontal headers.  Must be exactly
+		 * three columns, or this won't be unique.
+		 */
+		if (PQnfields(res) != 3)
+		{
+			pg_log_error("\\crosstabview: data column must be specified when query returns more than three columns");
+			goto error_return;
+		}
+
+		field_for_data = -1;
+		for (i = 0; i < PQnfields(res); i++)
+		{
+			if (i != field_for_rows && i != field_for_columns)
+			{
+				field_for_data = i;
+				break;
+			}
+		}
+		Assert(field_for_data >= 0);
+	}
+	else
+	{
+		field_for_data = indexOfColumn(pset.ctv_args[2], res);
+		if (field_for_data < 0)
+			goto error_return;
+	}
+
+	/* Process fourth optional arg (horizontal header sort column) */
+	if (pset.ctv_args[3] == NULL)
+		sort_field_for_columns = -1;	/* no sort column */
+	else
+	{
+		sort_field_for_columns = indexOfColumn(pset.ctv_args[3], res);
+		if (sort_field_for_columns < 0)
+			goto error_return;
+	}
+
+	/*
+	 * First part: accumulate the names that go into the vertical and
+	 * horizontal headers, each into an AVL binary tree to build the set of
+	 * DISTINCT values.
+	 */
+
+	for (rn = 0; rn < PQntuples(res); rn++)
+	{
+		char	   *val;
+		char	   *val1;
+
+		/* horizontal */
+		val = PQgetisnull(res, rn, field_for_columns) ? NULL :
+			PQgetvalue(res, rn, field_for_columns);
+		val1 = NULL;
+
+		if (sort_field_for_columns >= 0 &&
+			!PQgetisnull(res, rn, sort_field_for_columns))
+			val1 = PQgetvalue(res, rn, sort_field_for_columns);
+
+		avlMergeValue(&piv_columns, val, val1);
+
+		if (piv_columns.count > CROSSTABVIEW_MAX_COLUMNS)
+		{
+			pg_log_error("\\crosstabview: maximum number of columns (%d) exceeded",
+						 CROSSTABVIEW_MAX_COLUMNS);
+			goto error_return;
+		}
+
+		/* vertical */
+		val = PQgetisnull(res, rn, field_for_rows) ? NULL :
+			PQgetvalue(res, rn, field_for_rows);
+
+		avlMergeValue(&piv_rows, val, NULL);
+	}
+
+	/*
+	 * Second part: Generate sorted arrays from the AVL trees.
+	 */
+
+	num_columns = piv_columns.count;
+	num_rows = piv_rows.count;
+
+	array_columns = (pivot_field *)
+		pg_malloc(sizeof(pivot_field) * num_columns);
+
+	array_rows = (pivot_field *)
+		pg_malloc(sizeof(pivot_field) * num_rows);
+
+	avlCollectFields(&piv_columns, piv_columns.root, array_columns, 0);
+	avlCollectFields(&piv_rows, piv_rows.root, array_rows, 0);
+
+	/*
+	 * Third part: optionally, process the ranking data for the horizontal
+	 * header
+	 */
+	if (sort_field_for_columns >= 0)
+		rankSort(num_columns, array_columns);
+
+	/*
+	 * Fourth part: print the crosstab'ed result.
+	 */
+	retval = printCrosstab(res,
+						   num_columns, array_columns, field_for_columns,
+						   num_rows, array_rows, field_for_rows,
+						   field_for_data);
+
+error_return:
+	avlFree(&piv_columns, piv_columns.root);
+	avlFree(&piv_rows, piv_rows.root);
+	pg_free(array_columns);
+	pg_free(array_rows);
+
+	return retval;
+}
+
+/*
+ * Output the pivoted resultset with the printTable* functions.  Return true
+ * if successful, false otherwise.
+ */
+static bool
+printCrosstab(const PGresult *result,
+			  int num_columns, pivot_field *piv_columns, int field_for_columns,
+			  int num_rows, pivot_field *piv_rows, int field_for_rows,
+			  int field_for_data)
+{
+	printQueryOpt popt = pset.popt;
+	printTableContent cont;
+	int			i,
+				rn;
+	char		col_align;
+	int		   *horiz_map;
+	bool		retval = false;
+
+	printTableInit(&cont, &popt.topt, popt.title, num_columns + 1, num_rows);
+
+	/* Step 1: set target column names (horizontal header) */
+
+	/* The name of the first column is kept unchanged by the pivoting */
+	printTableAddHeader(&cont,
+						PQfname(result, field_for_rows),
+						false,
+						column_type_alignment(PQftype(result,
+													  field_for_rows)));
+
+	/*
+	 * To iterate over piv_columns[] by piv_columns[].rank, create a reverse
+	 * map associating each piv_columns[].rank to its index in piv_columns.
+	 * This avoids an O(N^2) loop later.
+	 */
+	horiz_map = (int *) pg_malloc(sizeof(int) * num_columns);
+	for (i = 0; i < num_columns; i++)
+		horiz_map[piv_columns[i].rank] = i;
+
+	/*
+	 * The display alignment depends on its PQftype().
+	 */
+	col_align = column_type_alignment(PQftype(result, field_for_data));
+
+	for (i = 0; i < num_columns; i++)
+	{
+		char	   *colname;
+
+		colname = piv_columns[horiz_map[i]].name ?
+			piv_columns[horiz_map[i]].name :
+			(popt.nullPrint ? popt.nullPrint : "");
+
+		printTableAddHeader(&cont, colname, false, col_align);
+	}
+	pg_free(horiz_map);
+
+	/* Step 2: set row names in the first output column (vertical header) */
+	for (i = 0; i < num_rows; i++)
+	{
+		int			k = piv_rows[i].rank;
+
+		cont.cells[k * (num_columns + 1)] = piv_rows[i].name ?
+			piv_rows[i].name :
+			(popt.nullPrint ? popt.nullPrint : "");
+	}
+	cont.cellsadded = num_rows * (num_columns + 1);
+
+	/*
+	 * Step 3: fill in the content cells.
+	 */
+	for (rn = 0; rn < PQntuples(result); rn++)
+	{
+		int			row_number;
+		int			col_number;
+		pivot_field *rp,
+				   *cp;
+		pivot_field elt;
+
+		/* Find target row */
+		if (!PQgetisnull(result, rn, field_for_rows))
+			elt.name = PQgetvalue(result, rn, field_for_rows);
+		else
+			elt.name = NULL;
+		rp = (pivot_field *) bsearch(&elt,
+									 piv_rows,
+									 num_rows,
+									 sizeof(pivot_field),
+									 pivotFieldCompare);
+		Assert(rp != NULL);
+		row_number = rp->rank;
+
+		/* Find target column */
+		if (!PQgetisnull(result, rn, field_for_columns))
+			elt.name = PQgetvalue(result, rn, field_for_columns);
+		else
+			elt.name = NULL;
+
+		cp = (pivot_field *) bsearch(&elt,
+									 piv_columns,
+									 num_columns,
+									 sizeof(pivot_field),
+									 pivotFieldCompare);
+		Assert(cp != NULL);
+		col_number = cp->rank;
+
+		/* Place value into cell */
+		if (col_number >= 0 && row_number >= 0)
+		{
+			int			idx;
+
+			/* index into the cont.cells array */
+			idx = 1 + col_number + row_number * (num_columns + 1);
+
+			/*
+			 * If the cell already contains a value, raise an error.
+			 */
+			if (cont.cells[idx] != NULL)
+			{
+				pg_log_error("\\crosstabview: query result contains multiple data values for row \"%s\", column \"%s\"",
+							 rp->name ? rp->name :
+							 (popt.nullPrint ? popt.nullPrint : "(null)"),
+							 cp->name ? cp->name :
+							 (popt.nullPrint ? popt.nullPrint : "(null)"));
+				goto error;
+			}
+
+			cont.cells[idx] = !PQgetisnull(result, rn, field_for_data) ?
+				PQgetvalue(result, rn, field_for_data) :
+				(popt.nullPrint ? popt.nullPrint : "");
+		}
+	}
+
+	/*
+	 * The non-initialized cells must be set to an empty string for the print
+	 * functions
+	 */
+	for (i = 0; i < cont.cellsadded; i++)
+	{
+		if (cont.cells[i] == NULL)
+			cont.cells[i] = "";
+	}
+
+	printTable(&cont, pset.queryFout, false, pset.logfile);
+	retval = true;
+
+error:
+	printTableCleanup(&cont);
+
+	return retval;
+}
+
+/*
+ * The avl* functions below provide a minimalistic implementation of AVL binary
+ * trees, to efficiently collect the distinct values that will form the horizontal
+ * and vertical headers. It only supports adding new values, no removal or even
+ * search.
+ */
+static void
+avlInit(avl_tree *tree)
+{
+	tree->end = (avl_node *) pg_malloc0(sizeof(avl_node));
+	tree->end->children[0] = tree->end->children[1] = tree->end;
+	tree->count = 0;
+	tree->root = tree->end;
+}
+
+/* Deallocate recursively an AVL tree, starting from node */
+static void
+avlFree(avl_tree *tree, avl_node *node)
+{
+	if (node->children[0] != tree->end)
+	{
+		avlFree(tree, node->children[0]);
+		pg_free(node->children[0]);
+	}
+	if (node->children[1] != tree->end)
+	{
+		avlFree(tree, node->children[1]);
+		pg_free(node->children[1]);
+	}
+	if (node == tree->root)
+	{
+		/* free the root separately as it's not child of anything */
+		if (node != tree->end)
+			pg_free(node);
+		/* free the tree->end struct only once and when all else is freed */
+		pg_free(tree->end);
+	}
+}
+
+/* Set the height to 1 plus the greatest of left and right heights */
+static void
+avlUpdateHeight(avl_node *n)
+{
+	n->height = 1 + (n->children[0]->height > n->children[1]->height ?
+					 n->children[0]->height :
+					 n->children[1]->height);
+}
+
+/* Rotate a subtree left (dir=0) or right (dir=1). Not recursive */
+static avl_node *
+avlRotate(avl_node **current, int dir)
+{
+	avl_node   *before = *current;
+	avl_node   *after = (*current)->children[dir];
+
+	*current = after;
+	before->children[dir] = after->children[!dir];
+	avlUpdateHeight(before);
+	after->children[!dir] = before;
+
+	return after;
+}
+
+static int
+avlBalance(avl_node *n)
+{
+	return n->children[0]->height - n->children[1]->height;
+}
+
+/*
+ * After an insertion, possibly rebalance the tree so that the left and right
+ * node heights don't differ by more than 1.
+ * May update *node.
+ */
+static void
+avlAdjustBalance(avl_tree *tree, avl_node **node)
+{
+	avl_node   *current = *node;
+	int			b = avlBalance(current) / 2;
+
+	if (b != 0)
+	{
+		int			dir = (1 - b) / 2;
+
+		if (avlBalance(current->children[dir]) == -b)
+			avlRotate(&current->children[dir], !dir);
+		current = avlRotate(node, dir);
+	}
+	if (current != tree->end)
+		avlUpdateHeight(current);
+}
+
+/*
+ * Insert a new value/field, starting from *node, reaching the correct position
+ * in the tree by recursion.  Possibly rebalance the tree and possibly update
+ * *node.  Do nothing if the value is already present in the tree.
+ */
+static void
+avlInsertNode(avl_tree *tree, avl_node **node, pivot_field field)
+{
+	avl_node   *current = *node;
+
+	if (current == tree->end)
+	{
+		avl_node   *new_node = (avl_node *)
+		pg_malloc(sizeof(avl_node));
+
+		new_node->height = 1;
+		new_node->field = field;
+		new_node->children[0] = new_node->children[1] = tree->end;
+		tree->count++;
+		*node = new_node;
+	}
+	else
+	{
+		int			cmp = pivotFieldCompare(&field, &current->field);
+
+		if (cmp != 0)
+		{
+			avlInsertNode(tree,
+						  cmp > 0 ? &current->children[1] : &current->children[0],
+						  field);
+			avlAdjustBalance(tree, node);
+		}
+	}
+}
+
+/* Insert the value into the AVL tree, if it does not preexist */
+static void
+avlMergeValue(avl_tree *tree, char *name, char *sort_value)
+{
+	pivot_field field;
+
+	field.name = name;
+	field.rank = tree->count;
+	field.sort_value = sort_value;
+	avlInsertNode(tree, &tree->root, field);
+}
+
+/*
+ * Recursively extract node values into the names array, in sorted order with a
+ * left-to-right tree traversal.
+ * Return the next candidate offset to write into the names array.
+ * fields[] must be preallocated to hold tree->count entries
+ */
+static int
+avlCollectFields(avl_tree *tree, avl_node *node, pivot_field *fields, int idx)
+{
+	if (node == tree->end)
+		return idx;
+
+	idx = avlCollectFields(tree, node->children[0], fields, idx);
+	fields[idx] = node->field;
+	return avlCollectFields(tree, node->children[1], fields, idx + 1);
+}
+
+static void
+rankSort(int num_columns, pivot_field *piv_columns)
+{
+	int		   *hmap;			/* [[offset in piv_columns, rank], ...for
+								 * every header entry] */
+	int			i;
+
+	hmap = (int *) pg_malloc(sizeof(int) * num_columns * 2);
+	for (i = 0; i < num_columns; i++)
+	{
+		char	   *val = piv_columns[i].sort_value;
+
+		/* ranking information is valid if non null and matches /^-?\d+$/ */
+		if (val &&
+			((*val == '-' &&
+			  strspn(val + 1, "0123456789") == strlen(val + 1)) ||
+			 strspn(val, "0123456789") == strlen(val)))
+		{
+			hmap[i * 2] = atoi(val);
+			hmap[i * 2 + 1] = i;
+		}
+		else
+		{
+			/* invalid rank information ignored (equivalent to rank 0) */
+			hmap[i * 2] = 0;
+			hmap[i * 2 + 1] = i;
+		}
+	}
+
+	qsort(hmap, num_columns, sizeof(int) * 2, rankCompare);
+
+	for (i = 0; i < num_columns; i++)
+	{
+		piv_columns[hmap[i * 2 + 1]].rank = i;
+	}
+
+	pg_free(hmap);
+}
+
+/*
+ * Look up a column reference, which can be either:
+ * - a number from 1 to PQnfields(res)
+ * - a column name matching one of PQfname(res,...)
+ *
+ * Returns zero-based column number, or -1 if not found or ambiguous.
+ *
+ * Note: may modify contents of "arg" string.
+ */
+static int
+indexOfColumn(char *arg, const PGresult *res)
+{
+	int			idx;
+
+	if (arg[0] && strspn(arg, "0123456789") == strlen(arg))
+	{
+		/* if arg contains only digits, it's a column number */
+		idx = atoi(arg) - 1;
+		if (idx < 0 || idx >= PQnfields(res))
+		{
+			pg_log_error("\\crosstabview: column number %d is out of range 1..%d",
+						 idx + 1, PQnfields(res));
+			return -1;
+		}
+	}
+	else
+	{
+		int			i;
+
+		/*
+		 * Dequote and downcase the column name.  By checking for all-digits
+		 * before doing this, we can ensure that a quoted name is treated as a
+		 * name even if it's all digits.
+		 */
+		dequote_downcase_identifier(arg, true, pset.encoding);
+
+		/* Now look for match(es) among res' column names */
+		idx = -1;
+		for (i = 0; i < PQnfields(res); i++)
+		{
+			if (strcmp(arg, PQfname(res, i)) == 0)
+			{
+				if (idx >= 0)
+				{
+					/* another idx was already found for the same name */
+					pg_log_error("\\crosstabview: ambiguous column name: \"%s\"", arg);
+					return -1;
+				}
+				idx = i;
+			}
+		}
+		if (idx == -1)
+		{
+			pg_log_error("\\crosstabview: column name not found: \"%s\"", arg);
+			return -1;
+		}
+	}
+
+	return idx;
+}
+
+/*
+ * Value comparator for vertical and horizontal headers
+ * used for deduplication only.
+ * - null values are considered equal
+ * - non-null < null
+ * - non-null values are compared with strcmp()
+ */
+static int
+pivotFieldCompare(const void *a, const void *b)
+{
+	const pivot_field *pa = (const pivot_field *) a;
+	const pivot_field *pb = (const pivot_field *) b;
+
+	/* test null values */
+	if (!pb->name)
+		return pa->name ? -1 : 0;
+	else if (!pa->name)
+		return 1;
+
+	/* non-null values */
+	return strcmp(pa->name, pb->name);
+}
+
+static int
+rankCompare(const void *a, const void *b)
+{
+	return *((const int *) a) - *((const int *) b);
+}