Hi,
Here's an updated patch that replaces sorted arrays by AVL binary trees
when gathering distinct values for the columns involved in the pivot.
The change is essential for large resultsets. For instance,
it allows to process a query like this (10 million rows x 10 columns):
select x,(random()*10)::int, (random()*1000)::int from
generate_series(1,10000000) as x
\crosstabview
which takes about 30 seconds to run and display on my machine with the
attached patch. That puts it seemingly in the same ballpark than
the equivalent test with the server-side crosstab().
With the previous iterations of the patch, this test would never end,
even with much smaller sets, as the execution time of the 1st step
grew exponentially with the number of distinct keys.
The exponential effect starts to be felt at about 10k values on my low-end
CPU,
and from there quickly becomes problematic.
As a client-side display feature, processing millions of rows like in
the query above does not necessarily make sense, it's pushing the
envelope, but stalling way below 100k rows felt lame, so I'm happy to get
rid of that limitation.
However, there is another one. The above example does not need or request
an additional sort step, but if it did, sorting more than 65535 entries in
the vertical header would error out, because values are shipped as
parameters to PQexecParams(), which only accepts that much.
To avoid the problem, when the rows in the output "grid" exceed 2^16 and
they need to be sorted, the user must let the sort being driven by ORDER BY
beforehand in the query, knowing that the pivot will keep the original
ordering intact in the vertical header.
I'm still thinking about extending this based on Pavel's diff for the
"label" column, so that
\crosstabview [+|-]colV[:colSortH] [+|-]colH[:colSortH]
would mean to use colV/H as grid headers but sort them according
to colSortV/H.
I prefer that syntax over adding more parameters, and also I'd like
to have it work in both V and H directions.
Aside from the AVL trees, there are a few other minor changes in that
patch:
- move non-exportable structs from the .h to the .c
- move code in common.c to respect alphabetical ordering
- if vertical sort is requested, add explicit check against more than 65535
params instead of letting the sort query fail
- report all failure cases of the sort query
- rename sortColumns to serverSort and use less the term "columns" in
comments and variables.
Best regards,
--
Daniel Vérité
PostgreSQL-powered mailer: http://www.manitou-mail.org
Twitter: @DanielVerite
diff --git a/doc/src/sgml/ref/psql-ref.sgml b/doc/src/sgml/ref/psql-ref.sgml
index 6d0cb3d..563324d 100644
--- a/doc/src/sgml/ref/psql-ref.sgml
+++ b/doc/src/sgml/ref/psql-ref.sgml
@@ -2485,6 +2485,95 @@ lo_import 152801
</listitem>
</varlistentry>
+ <varlistentry>
+ <term><literal>\crosstabview [ [-|+]<replaceable class="parameter">colV</replaceable> [-|+]<replaceable class="parameter">colH</replaceable> ] </literal></term>
+ <listitem>
+ <para>
+ Execute the current query buffer (like <literal>\g</literal>) and shows the results
+ inside a crosstab grid. The output column <replaceable class="parameter">colV</replaceable>
+ becomes a vertical header and the output column
+ <replaceable class="parameter">colH</replaceable> becomes a horizontal header.
+ The results for the other output columns are projected inside the grid.
+ </para>
+
+ <para>
+ <replaceable class="parameter">colV</replaceable>
+ and <replaceable class="parameter">colH</replaceable> can indicate a
+ column position (starting at 1), or a column name. Normal case folding
+ and quoting rules apply on column names. By default,
+ <replaceable class="parameter">colV</replaceable> is column 1
+ and <replaceable class="parameter">colH</replaceable> is column 2.
+ A query having only one output column cannot be viewed in crosstab, and
+ <replaceable class="parameter">colH</replaceable> must differ from
+ <replaceable class="parameter">colV</replaceable>.
+ </para>
+
+ <para>
+ The vertical header, displayed as the leftmost column,
+ contains the set of all distinct values found in
+ column <replaceable class="parameter">colV</replaceable>, in the order
+ of their first appearance in the query results.
+ </para>
+ <para>
+ The horizontal header, displayed as the first row,
+ contains the set of all distinct non-null values found in
+ column <replaceable class="parameter">colH</replaceable>. They come
+ by default in their order of appearance in the query results, or in ascending
+ order if a plus (+) sign precedes <replaceable class="parameter">colH</replaceable>,
+ or in descending order if it's a minus (-) sign.
+ </para>
+
+ <para>
+ The query results being tuples of <literal>N</literal> columns
+ (including <replaceable class="parameter">colV</replaceable> and
+ <replaceable class="parameter">colH</replaceable>),
+ for each distinct value <literal>x</literal> of
+ <replaceable class="parameter">colH</replaceable>
+ and each distinct value <literal>y</literal> of
+ <replaceable class="parameter">colV</replaceable>,
+ a cell located at the intersection <literal>(x,y)</literal> in the grid
+ has contents determined by these rules:
+ <itemizedlist>
+ <listitem>
+ <para>
+ if there is no corresponding row in the results such that the value
+ for <replaceable class="parameter">colH</replaceable>
+ is <literal>x</literal> and the value
+ for <replaceable class="parameter">colV</replaceable>
+ is <literal>y</literal>, the cell is empty.
+ </para>
+ </listitem>
+
+ <listitem>
+ <para>
+ if there is exactly one row such that the value
+ for <replaceable class="parameter">colH</replaceable>
+ is <literal>x</literal> and the value
+ for <replaceable class="parameter">colV</replaceable>
+ is <literal>y</literal>, then the <literal>N-2</literal> other
+ columns are displayed in the cell, separated between each other by
+ a space character if needed.
+
+ If <literal>N=2</literal>, the letter <literal>X</literal> is displayed in the cell as
+ if a virtual third column contained that character.
+ </para>
+ </listitem>
+
+ <listitem>
+ <para>
+ if there are several corresponding rows, the behavior is identical to one row
+ except that the values coming from different rows are stacked
+ vertically, rows being separated by newline characters inside
+ the same cell.
+ </para>
+ </listitem>
+
+ </itemizedlist>
+ </para>
+
+ </listitem>
+ </varlistentry>
+
<varlistentry>
<term><literal>\s [ <replaceable class="parameter">filename</replaceable> ]</literal></term>
diff --git a/src/bin/psql/Makefile b/src/bin/psql/Makefile
index f1336d5..9cb0c4a 100644
--- a/src/bin/psql/Makefile
+++ b/src/bin/psql/Makefile
@@ -23,7 +23,7 @@ override CPPFLAGS := -I. -I$(srcdir) -I$(libpq_srcdir) -I$(top_srcdir)/src/bin/p
OBJS= command.o common.o help.o input.o stringutils.o mainloop.o copy.o \
startup.o prompt.o variables.o large_obj.o print.o describe.o \
tab-complete.o mbprint.o dumputils.o keywords.o kwlookup.o \
- sql_help.o \
+ sql_help.o crosstabview.o \
$(WIN32RES)
diff --git a/src/bin/psql/command.c b/src/bin/psql/command.c
index cf6876b..b5ec8af 100644
--- a/src/bin/psql/command.c
+++ b/src/bin/psql/command.c
@@ -39,6 +39,7 @@
#include "common.h"
#include "copy.h"
+#include "crosstabview.h"
#include "describe.h"
#include "help.h"
#include "input.h"
@@ -364,6 +365,34 @@ exec_command(const char *cmd,
else if (strcmp(cmd, "copyright") == 0)
print_copyright();
+ /* \crosstabview -- execute a query and display results in crosstab */
+ else if (strcmp(cmd, "crosstabview") == 0)
+ {
+ char *opt1,
+ *opt2;
+
+ opt1 = psql_scan_slash_option(scan_state,
+ OT_NORMAL, NULL, false);
+ opt2 = psql_scan_slash_option(scan_state,
+ OT_NORMAL, NULL, false);
+
+ if (opt1 && !opt2)
+ {
+ psql_error(_("\\%s: missing second argument\n"), cmd);
+ success = false;
+ }
+ else
+ {
+ pset.crosstabview_col_V = opt1 ? pg_strdup(opt1): NULL;
+ pset.crosstabview_col_H = opt2 ? pg_strdup(opt2): NULL;
+ pset.crosstabview_output = true;
+ status = PSQL_CMD_SEND;
+ }
+
+ free(opt1);
+ free(opt2);
+ }
+
/* \d* commands */
else if (cmd[0] == 'd')
{
diff --git a/src/bin/psql/common.c b/src/bin/psql/common.c
index a287eee..f5db53d 100644
--- a/src/bin/psql/common.c
+++ b/src/bin/psql/common.c
@@ -24,6 +24,7 @@
#include "command.h"
#include "copy.h"
#include "mbprint.h"
+#include "crosstabview.h"
static bool ExecQueryUsingCursor(const char *query, double *elapsed_msec);
@@ -906,6 +907,8 @@ PrintQueryResults(PGresult *results)
/* store or print the data ... */
if (pset.gset_prefix)
success = StoreQueryTuple(results);
+ else if (pset.crosstabview_output)
+ success = PrintResultsInCrossTab(results);
else
success = PrintQueryTuples(results);
/* if it's INSERT/UPDATE/DELETE RETURNING, also print status */
@@ -1192,6 +1195,18 @@ sendquery_cleanup:
pset.gset_prefix = NULL;
}
+ /* reset \crosstabview settings */
+ pset.crosstabview_output = false;
+ if (pset.crosstabview_col_V)
+ {
+ free(pset.crosstabview_col_V);
+ pset.crosstabview_col_V = NULL;
+ }
+ if (pset.crosstabview_col_H)
+ {
+ free(pset.crosstabview_col_H);
+ pset.crosstabview_col_H = NULL;
+ }
return OK;
}
@@ -1354,7 +1369,25 @@ ExecQueryUsingCursor(const char *query, double *elapsed_msec)
is_pager = true;
}
- printQuery(results, &my_popt, fout, is_pager, pset.logfile);
+ if (pset.crosstabview_output)
+ {
+ if (ntuples < fetch_count)
+ PrintResultsInCrossTab(results);
+ else
+ {
+ /*
+ crosstabview is denied if the whole set of rows is not
+ guaranteed to be fetched in the first iteration, because
+ it's expected in memory as a single PGresult structure.
+ */
+ psql_error("\\crosstabview must be used with less than FETCH_COUNT (%d) rows\n",
+ fetch_count);
+ PQclear(results);
+ break;
+ }
+ }
+ else
+ printQuery(results, &my_popt, fout, is_pager, pset.logfile);
PQclear(results);
diff --git a/src/bin/psql/crosstabview.c b/src/bin/psql/crosstabview.c
new file mode 100644
index 0000000..82f3e2e
--- /dev/null
+++ b/src/bin/psql/crosstabview.c
@@ -0,0 +1,845 @@
+/*
+ * psql - the PostgreSQL interactive terminal
+ *
+ * Copyright (c) 2000-2015, PostgreSQL Global Development Group
+ *
+ * src/bin/psql/crosstabview.c
+ */
+
+#include "common.h"
+#include "crosstabview.h"
+#include "pqexpbuffer.h"
+#include "settings.h"
+#include <string.h>
+
+/*
+ * Value/position from the resultset that goes into the horizontal or vertical
+ * crosstabview header.
+ */
+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)
+ */
+ char *name;
+
+ /*
+ * Rank of this value, starting at 0. Initially, it's the relative position
+ * of the first appearance of this value 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
+ * After all values have been gathered, ranks may be updated by serverSort()
+ */
+ int rank;
+};
+
+/* Node in avl_tree */
+struct avl_node
+{
+ /* Node contents */
+ struct 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 *childs[2];
+};
+
+/*
+ * Control structure for the AVL tree (binary search tree kept
+ * balanced with the AVL algorithm)
+ */
+struct avl_tree
+{
+ int count; /* Total number of nodes */
+ int freecnt;
+ struct avl_node *root; /* root of the tree */
+ struct avl_node *end; /* Immutable dereferenceable empty tree */
+};
+
+/*
+ * 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(struct avl_tree *tree)
+{
+ tree->end = (struct avl_node*) pg_malloc0(sizeof(struct avl_node));
+ tree->end->childs[0] = tree->end->childs[1] = tree->end;
+ tree->count = 0;
+ tree->root = tree->end;
+ tree->freecnt=0;
+}
+
+/* Deallocate recursively an AVL tree, starting from node */
+static void
+avlFree(struct avl_tree* tree, struct avl_node* node)
+{
+ if (node->childs[0] != tree->end)
+ {
+ avlFree(tree, node->childs[0]);
+ pg_free(node->childs[0]);
+ }
+ if (node->childs[1] != tree->end)
+ {
+ avlFree(tree, node->childs[1]);
+ pg_free(node->childs[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(struct avl_node *n)
+{
+ n->height = 1 + (n->childs[0]->height > n->childs[1]->height ?
+ n->childs[0]->height:
+ n->childs[1]->height);
+}
+
+/* Rotate a subtree left (dir=0) or right (dir=1). Not recursive */
+static struct avl_node*
+avlRotate(struct avl_node **current, int dir)
+{
+ struct avl_node *before = *current;
+ struct avl_node *after = (*current)->childs[dir];
+
+ *current = after;
+ before->childs[dir] = after->childs[!dir];
+ avlUpdateHeight(before);
+ after->childs[!dir] = before;
+
+ return after;
+}
+
+static int
+avlBalance(struct avl_node *n)
+{
+ return n->childs[0]->height - n->childs[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(struct avl_tree *tree, struct avl_node **node)
+{
+ struct avl_node *current = *node;
+ int b = avlBalance(current)/2;
+ if (b != 0)
+ {
+ int dir = (1 - b)/2;
+ if (avlBalance(current->childs[dir]) == -b)
+ avlRotate(¤t->childs[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(struct avl_tree* tree,
+ struct avl_node **node,
+ struct pivot_field field)
+{
+ struct avl_node *current = *node;
+
+ if (current == tree->end)
+ {
+ struct avl_node * new_node = (struct avl_node*)
+ pg_malloc(sizeof(struct avl_node));
+ new_node->height = 1;
+ new_node->field = field;
+ new_node->childs[0] = new_node->childs[1] = tree->end;
+ tree->count++;
+ *node = new_node;
+ }
+ else
+ {
+ int cmp = strcmp(field.name, current->field.name);
+ if (cmp != 0)
+ {
+ avlInsertNode(tree,
+ cmp > 0 ? ¤t->childs[1] : ¤t->childs[0],
+ field);
+ avlAdjustBalance(tree, node);
+ }
+ }
+}
+
+/* Insert the value into the AVL tree, if it does not preexist */
+static void
+avlMergeValue(struct avl_tree* tree, char* name)
+{
+ struct pivot_field field;
+ field.name = name;
+ field.rank = tree->count;
+ 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(struct avl_tree* tree,
+ struct avl_node* node,
+ struct pivot_field* fields,
+ int idx)
+{
+ if (node == tree->end)
+ return idx;
+ idx = avlCollectFields(tree, node->childs[0], fields, idx);
+ fields[idx] = node->field;
+ return avlCollectFields(tree, node->childs[1], fields, idx+1);
+}
+
+/*
+ * Send a query to sort the list of values in a horizontal or vertical
+ * crosstabview header, and update every source[].rank field with the
+ * new relative position of each value.
+ * coltype is the type's OID for the column from which the values come.
+ * direction is -1 for descending, 1 for ascending, 0 for no sort.
+ */
+static bool
+serverSort(Oid coltype, struct pivot_field *source, int nb_values, int direction)
+{
+ bool retval = false;
+ PGresult *res = NULL;
+ PQExpBufferData query;
+ int i;
+ Oid *param_types;
+ const char** param_values;
+ int* param_lengths;
+ int* param_formats;
+
+ if (nb_values < 2 || direction==0)
+ return true; /* nothing to sort */
+
+ param_types = (Oid*) pg_malloc(nb_values*sizeof(Oid));
+ param_values = (const char**) pg_malloc(nb_values*sizeof(char*));
+ param_lengths = (int*) pg_malloc(nb_values*sizeof(int));
+ param_formats = (int*) pg_malloc(nb_values*sizeof(int));
+
+ initPQExpBuffer(&query);
+
+ /*
+ * The query returns the original position of each value in our list,
+ * ordered by its new position. The value itself is not returned.
+ */
+ appendPQExpBufferStr(&query, "SELECT n FROM (VALUES");
+
+ for (i=1; i <= nb_values; i++)
+ {
+ if (i < nb_values)
+ appendPQExpBuffer(&query, "($%d,%d),", i, i);
+ else
+ {
+ appendPQExpBuffer(&query, "($%d,%d)) AS l(x,n) ORDER BY x", i, i);
+ if (direction < 0)
+ appendPQExpBufferStr(&query, " DESC");
+ }
+
+ param_types[i-1] = coltype;
+ param_values[i-1] = source[i-1].name;
+ param_lengths[i-1] = strlen(source[i-1].name);
+ param_formats[i-1] = 0;
+ }
+
+ res = PQexecParams(pset.db,
+ query.data,
+ nb_values,
+ param_types,
+ param_values,
+ param_lengths,
+ param_formats,
+ 0);
+
+ if (res && PQresultStatus(res) == PGRES_TUPLES_OK)
+ {
+ for (i=0; i < PQntuples(res); i++)
+ {
+ int old_pos = atoi(PQgetvalue(res, i, 0));
+
+ if (old_pos < 1 || old_pos > nb_values || i >= nb_values)
+ {
+ /*
+ * A position outside of the range is normally impossible.
+ * If this happens, we're facing a malfunctioning or hostile
+ * server or middleware.
+ */
+ psql_error(_("Unexpected rank when sorting crosstabview headers"));
+ goto cleanup;
+ }
+ else
+ {
+ source[old_pos-1].rank = i;
+ }
+ }
+ }
+ else
+ {
+ psql_error(_("Query error when sorting crosstabview header: %s"),
+ PQerrorMessage(pset.db));
+ goto cleanup;
+ }
+
+ retval = true;
+
+cleanup:
+ termPQExpBuffer(&query);
+ if (res)
+ PQclear(res);
+ pg_free(param_types);
+ pg_free(param_values);
+ pg_free(param_lengths);
+ pg_free(param_formats);
+ return retval;
+}
+
+/* for bsearch() inside a sorted array of struct pivot_field */
+static int
+headerCompare(const void *a, const void *b)
+{
+ return strcmp( ((struct pivot_field*)a)->name,
+ ((struct pivot_field*)b)->name);
+}
+
+
+static void
+printCrosstab(const PGresult *results,
+ int num_columns,
+ struct pivot_field *piv_columns,
+ int field_for_columns,
+ int num_rows,
+ struct pivot_field *piv_rows,
+ int field_for_rows)
+{
+ printQueryOpt popt = pset.popt;
+ printTableContent cont;
+ int i, j, rn;
+ char col_align = 'l'; /* alignment for values inside the grid */
+ int* horiz_map; /* map indices from sorted horizontal headers to piv_columns */
+ char** allocated_cells; /* Pointers for cell contents that are allocated
+ * in this function, when cells cannot simply point to
+ * PQgetvalue(results, ...) */
+
+ 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(results, field_for_rows),
+ false,
+ column_type_alignment(PQftype(results, 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;
+ }
+
+ /*
+ * In the case of 3 output columns, the contents in the cells are exactly
+ * the contents of the "value" column (3rd column by default), so their
+ * alignment is determined by PQftype(). Otherwise the contents are
+ * made-up strings, so the alignment is 'l'
+ */
+ if (PQnfields(results) == 3)
+ {
+ int colnum; /* column placed inside the grid */
+ /*
+ * find colnum in the permutations of (0,1,2) where colnum is
+ * neither field_for_rows nor field_for_columns
+ */
+ switch (field_for_rows)
+ {
+ case 0:
+ colnum = (field_for_columns == 1) ? 2 : 1;
+ break;
+ case 1:
+ colnum = (field_for_columns == 0) ? 2: 0;
+ break;
+ default: /* should be always 2 */
+ colnum = (field_for_columns == 0) ? 1: 0;
+ break;
+ }
+ col_align = column_type_alignment(PQftype(results, colnum));
+ }
+ else
+ col_align = 'l';
+
+ for (i = 0; i < num_columns; i++)
+ {
+ printTableAddHeader(&cont,
+ piv_columns[horiz_map[i]].name,
+ 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;
+ /* Initialize all cells inside the grid to an empty value */
+ for (j = 0; j < num_columns; j++)
+ cont.cells[k*(num_columns+1)+j+1] = "";
+ }
+ cont.cellsadded = num_rows * (num_columns+1);
+
+ allocated_cells = (char**) pg_malloc0(num_rows * num_columns * sizeof(char*));
+
+ /* Step 3: set all the cells "inside the grid" */
+ for (rn = 0; rn < PQntuples(results); rn++)
+ {
+ char* row_name;
+ char* col_name;
+ int row_number;
+ int col_number;
+ struct pivot_field *p;
+
+ row_number = col_number = -1;
+ /* Find target row */
+ if (!PQgetisnull(results, rn, field_for_rows))
+ {
+ row_name = PQgetvalue(results, rn, field_for_rows);
+ p = (struct pivot_field*) bsearch(&row_name,
+ piv_rows,
+ num_rows,
+ sizeof(struct pivot_field),
+ headerCompare);
+ if (p)
+ row_number = p->rank;
+ }
+
+ /* Find target column */
+ if (!PQgetisnull(results, rn, field_for_columns))
+ {
+ col_name = PQgetvalue(results, rn, field_for_columns);
+ p = (struct pivot_field*) bsearch(&col_name,
+ piv_columns,
+ num_columns,
+ sizeof(struct pivot_field),
+ headerCompare);
+ if (p)
+ col_number = p->rank;
+ }
+
+ /* Place value into cell */
+ if (col_number>=0 && row_number>=0)
+ {
+ int idx = 1 + col_number + row_number*(num_columns+1);
+ int src_col = 0; /* column number in source result */
+ int k = 0;
+
+ do {
+ char *content;
+
+ if (PQnfields(results) == 2)
+ {
+ /*
+ special case: when the source has only 2 columns, use a
+ X (cross/checkmark) for the cell content, and set
+ src_col to a virtual additional column.
+ */
+ content = "X";
+ src_col = 3;
+ }
+ else if (src_col == field_for_rows || src_col == field_for_columns)
+ {
+ /*
+ The source values that produce headers are not processed
+ in this loop, only the values that end up inside the grid.
+ */
+ src_col++;
+ continue;
+ }
+ else
+ {
+ content = (!PQgetisnull(results, rn, src_col)) ?
+ PQgetvalue(results, rn, src_col) :
+ (popt.nullPrint ? popt.nullPrint : "");
+ }
+
+ if (cont.cells[idx] != NULL && cont.cells[idx][0] != '\0')
+ {
+ /*
+ * Multiple values for the same (row,col) are projected
+ * into the same cell. When this happens, separate the
+ * previous content of the cell from the new value by a
+ * newline.
+ */
+ int content_size =
+ strlen(cont.cells[idx])
+ + 2 /* room for [CR],LF or space */
+ + strlen(content)
+ + 1; /* '\0' */
+ char *new_content;
+
+ /*
+ * idx2 is an index into allocated_cells. It differs from
+ * idx (index into cont.cells), because vertical and
+ * horizontal headers are included in `cont.cells` but
+ * excluded from allocated_cells.
+ */
+ int idx2 = (row_number * num_columns) + col_number;
+
+ if (allocated_cells[idx2] != NULL)
+ {
+ new_content = pg_realloc(allocated_cells[idx2], content_size);
+ }
+ else
+ {
+ /*
+ * At this point, cont.cells[idx] still contains a
+ * PQgetvalue() pointer. Just after, it will contain
+ * a new pointer maintained in allocated_cells[], and
+ * freed at the end of this function.
+ */
+ new_content = pg_malloc(content_size);
+ strcpy(new_content, cont.cells[idx]);
+ }
+ cont.cells[idx] = new_content;
+ allocated_cells[idx2] = new_content;
+
+ /*
+ * Contents that are on adjacent columns in the source results get
+ * separated by one space in the target.
+ * Contents that are on different rows in the source get
+ * separated by newlines in the target.
+ */
+ if (k==0)
+ strcat(new_content, "\n");
+ else
+ strcat(new_content, " ");
+ strcat(new_content, content);
+ }
+ else
+ {
+ cont.cells[idx] = content;
+ }
+ k++;
+ src_col++;
+ } while (src_col < PQnfields(results));
+ }
+ }
+
+ printTable(&cont, pset.queryFout, false, pset.logfile);
+ printTableCleanup(&cont);
+
+
+ for (i=0; i < num_rows * num_columns; i++)
+ {
+ if (allocated_cells[i] != NULL)
+ pg_free(allocated_cells[i]);
+ }
+
+ pg_free(allocated_cells);
+}
+
+
+/*
+ * Compare a user-supplied argument against a field name obtained by PQfname(),
+ * which is already case-folded.
+ * If arg is not enclosed in double quotes, pg_strcasecmp applies, otherwise
+ * do a case-sensitive comparison with these rules:
+ * - double quotes enclosing 'arg' are filtered out
+ * - double quotes inside 'arg' are expected to be doubled
+ */
+static int
+fieldNameCmp(const char* arg, const char* fieldname)
+{
+ const unsigned char* p = (const unsigned char*) arg;
+ const unsigned char* f = (const unsigned char*) fieldname;
+ unsigned char c;
+
+ if (*p++ != '"')
+ return pg_strcasecmp(arg, fieldname);
+
+ while ((c=*p++))
+ {
+ if (c=='"')
+ {
+ if (*p=='"')
+ p++; /* skip second quote and continue */
+ else if (*p=='\0')
+ return *p-*f; /* p finishes before f or is identical */
+
+ }
+ if (*f=='\0')
+ return 1; /* f finishes before p */
+ if (c!=*f)
+ return c-*f;
+ f++;
+ }
+ return (*f=='\0') ? 0 : 1;
+}
+
+
+/*
+ * arg can be a number or a column name, possibly quoted (like in an ORDER BY clause)
+ * Returns:
+ * on success, the 0-based index of the column
+ * or -1 if the column number or name is not found in the result's structure,
+ * or if it's ambiguous (arg corresponding to several columns)
+ */
+static int
+indexOfColumn(const char* arg, PGresult* res)
+{
+ int idx;
+
+ if (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))
+ {
+ psql_error(_("Invalid column number: %s\n"), arg);
+ return -1;
+ }
+ }
+ else
+ {
+ int i;
+ idx = -1;
+ for (i=0; i < PQnfields(res); i++)
+ {
+ if (fieldNameCmp(arg, PQfname(res, i)) == 0)
+ {
+ if (idx>=0)
+ {
+ /* if another idx was already found for the same name */
+ psql_error(_("Ambiguous column name: %s\n"), arg);
+ return -1;
+ }
+ idx = i;
+ }
+ }
+ if (idx == -1)
+ {
+ psql_error(_("Invalid column name: %s\n"), arg);
+ return -1;
+ }
+ }
+ return idx;
+}
+
+/*
+ * Main function.
+ * Process the data from *res according the display options in pset (global),
+ * to generate the horizontal and vertical headers contents,
+ * then call printCrosstab() for the actual output.
+ */
+bool
+PrintResultsInCrossTab(PGresult* res)
+{
+ /* [-|+]COLV or null */
+ const char* opt_field_for_rows = pset.crosstabview_col_V;
+ /* [-|+]COLH or null */
+ const char* opt_field_for_columns = pset.crosstabview_col_H;
+ int rn;
+ struct avl_tree piv_columns;
+ struct avl_tree piv_rows;
+ struct pivot_field* array_columns = NULL;
+ struct pivot_field* array_rows = NULL;
+ int num_columns = 0;
+ int num_rows = 0;
+ bool retval = false;
+ int columns_sort_direction = 0; /* 1:ascending, 0:none, -1:descending */
+ int rows_sort_direction = 0; /* 1:ascending, 0:none, -1:descending */
+
+ /* 0-based index of the field whose distinct values will become COLUMN headers */
+ int field_for_columns;
+
+ /* 0-based index of the field whose distinct values will become ROW headers */
+ int field_for_rows;
+
+ avlInit(&piv_rows);
+ avlInit(&piv_columns);
+
+ if (PQresultStatus(res) != PGRES_TUPLES_OK)
+ goto error_return;
+
+ if (PQnfields(res) < 2)
+ {
+ psql_error(_("The query must return at least two columns to be shown in crosstab\n"));
+ goto error_return;
+ }
+
+ /* field for rows in crosstab */
+ if (opt_field_for_rows == NULL)
+ field_for_rows = 0;
+ else
+ {
+ if (*opt_field_for_rows == '-')
+ {
+ rows_sort_direction = -1;
+ opt_field_for_rows++;
+ }
+ else if (*opt_field_for_rows == '+')
+ {
+ rows_sort_direction = 1;
+ opt_field_for_rows++;
+ }
+ field_for_rows = indexOfColumn(opt_field_for_rows, res);
+ if (field_for_rows < 0)
+ goto error_return;
+ }
+
+ if (field_for_rows < 0)
+ goto error_return;
+
+ /* field for columns in crosstab */
+ if (opt_field_for_columns == NULL)
+ field_for_columns = 1;
+ else
+ {
+ /*
+ * descending sort is requested if the column reference is
+ * preceded with a minus sign
+ */
+ if (*opt_field_for_columns == '-')
+ {
+ columns_sort_direction = -1;
+ opt_field_for_columns++;
+ }
+ else if (*opt_field_for_columns == '+')
+ {
+ columns_sort_direction = 1;
+ opt_field_for_columns++;
+ }
+ field_for_columns = indexOfColumn(opt_field_for_columns, res);
+ if (field_for_columns < 0)
+ goto error_return;
+ }
+
+
+ if (field_for_columns == field_for_rows)
+ {
+ psql_error(_("The same column cannot be used for both vertical and horizontal headers\n"));
+ 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++)
+ {
+ if (!PQgetisnull(res, rn, field_for_rows))
+ {
+ avlMergeValue(&piv_rows,
+ PQgetvalue(res, rn, field_for_rows));
+ if (rows_sort_direction!=0 && piv_rows.count > 65535)
+ {
+ psql_error(_("Maximum number of rows to sort (65535) exceeded for the vertical header\n"));
+ goto error_return;
+ }
+ }
+
+ if (!PQgetisnull(res, rn, field_for_columns))
+ {
+ avlMergeValue(&piv_columns,
+ PQgetvalue(res, rn, field_for_columns));
+ if (piv_columns.count > 1600)
+ {
+ psql_error(_("Maximum number of columns (1600) exceeded\n"));
+ goto error_return;
+ }
+ }
+ }
+
+
+ /*
+ * Second part: Generate sorted arrays from the AVL trees.
+ */
+
+ num_columns = piv_columns.count;
+ num_rows = piv_rows.count;
+
+ array_columns = (struct pivot_field*) pg_malloc(sizeof(struct pivot_field) * num_columns);
+ array_rows = (struct pivot_field*) pg_malloc(sizeof(struct pivot_field) * num_rows);
+
+ avlCollectFields(&piv_columns, piv_columns.root, array_columns, 0);
+ avlCollectFields(&piv_rows, piv_rows.root, array_rows, 0);
+
+ /*
+ * Third part: sort (by server-side query) the horizontal and/or vertical
+ * header when applicable.
+ */
+ if (columns_sort_direction != 0)
+ {
+ if (!serverSort(PQftype(res, field_for_columns),
+ array_columns,
+ num_columns,
+ columns_sort_direction))
+ goto error_return;
+ }
+
+ if (rows_sort_direction != 0)
+ {
+ if (!serverSort(PQftype(res, field_for_rows),
+ array_rows,
+ num_rows,
+ rows_sort_direction))
+ goto error_return;
+ }
+
+ /*
+ * Fourth part: print the crosstab'ed results.
+ */
+ printCrosstab(res,
+ num_columns,
+ array_columns,
+ field_for_columns,
+ num_rows,
+ array_rows,
+ field_for_rows);
+
+ retval = true;
+
+error_return:
+ avlFree(&piv_columns, piv_columns.root);
+ avlFree(&piv_rows, piv_rows.root);
+ pg_free(array_columns);
+ pg_free(array_rows);
+
+ return retval;
+}
diff --git a/src/bin/psql/crosstabview.h b/src/bin/psql/crosstabview.h
new file mode 100644
index 0000000..ccbfa02
--- /dev/null
+++ b/src/bin/psql/crosstabview.h
@@ -0,0 +1,14 @@
+/*
+ * psql - the PostgreSQL interactive terminal
+ *
+ * Copyright (c) 2000-2015, PostgreSQL Global Development Group
+ *
+ * src/bin/psql/crosstabview.h
+ */
+
+#ifndef CROSSTABVIEW_H
+#define CROSSTABVIEW_H
+
+/* prototypes */
+extern bool PrintResultsInCrossTab(PGresult *res);
+#endif /* CROSSTABVIEW_H */
diff --git a/src/bin/psql/help.c b/src/bin/psql/help.c
index 5f240be..3924046 100644
--- a/src/bin/psql/help.c
+++ b/src/bin/psql/help.c
@@ -175,6 +175,7 @@ slashUsage(unsigned short int pager)
fprintf(output, _(" \\g [FILE] or ; execute query (and send results to file or |pipe)\n"));
fprintf(output, _(" \\gset [PREFIX] execute query and store results in psql variables\n"));
fprintf(output, _(" \\q quit psql\n"));
+ fprintf(output, _(" \\crosstabview [V H] execute query and display results in crosstab\n"));
fprintf(output, _(" \\watch [SEC] execute query every SEC seconds\n"));
fprintf(output, "\n");
diff --git a/src/bin/psql/print.c b/src/bin/psql/print.c
index 05d4b31..b2f8c2b 100644
--- a/src/bin/psql/print.c
+++ b/src/bin/psql/print.c
@@ -3291,30 +3291,9 @@ printQuery(const PGresult *result, const printQueryOpt *opt,
for (i = 0; i < cont.ncolumns; i++)
{
- char align;
- Oid ftype = PQftype(result, i);
-
- switch (ftype)
- {
- case INT2OID:
- case INT4OID:
- case INT8OID:
- case FLOAT4OID:
- case FLOAT8OID:
- case NUMERICOID:
- case OIDOID:
- case XIDOID:
- case CIDOID:
- case CASHOID:
- align = 'r';
- break;
- default:
- align = 'l';
- break;
- }
-
printTableAddHeader(&cont, PQfname(result, i),
- opt->translate_header, align);
+ opt->translate_header,
+ column_type_alignment(PQftype(result, i)));
}
/* set cells */
@@ -3356,6 +3335,31 @@ printQuery(const PGresult *result, const printQueryOpt *opt,
printTableCleanup(&cont);
}
+char
+column_type_alignment(Oid ftype)
+{
+ char align;
+
+ switch (ftype)
+ {
+ case INT2OID:
+ case INT4OID:
+ case INT8OID:
+ case FLOAT4OID:
+ case FLOAT8OID:
+ case NUMERICOID:
+ case OIDOID:
+ case XIDOID:
+ case CIDOID:
+ case CASHOID:
+ align = 'r';
+ break;
+ default:
+ align = 'l';
+ break;
+ }
+ return align;
+}
void
setDecimalLocale(void)
diff --git a/src/bin/psql/print.h b/src/bin/psql/print.h
index fd565984..218b185 100644
--- a/src/bin/psql/print.h
+++ b/src/bin/psql/print.h
@@ -175,7 +175,7 @@ extern FILE *PageOutput(int lines, const printTableOpt *topt);
extern void ClosePager(FILE *pagerpipe);
extern void html_escaped_print(const char *in, FILE *fout);
-
+extern char column_type_alignment(Oid);
extern void printTableInit(printTableContent *const content,
const printTableOpt *opt, const char *title,
const int ncolumns, const int nrows);
diff --git a/src/bin/psql/settings.h b/src/bin/psql/settings.h
index 1885bb1..6069024 100644
--- a/src/bin/psql/settings.h
+++ b/src/bin/psql/settings.h
@@ -90,6 +90,9 @@ typedef struct _psqlSettings
char *gfname; /* one-shot file output argument for \g */
char *gset_prefix; /* one-shot prefix argument for \gset */
+ bool crosstabview_output; /* one-shot request to print results in crosstab */
+ char *crosstabview_col_V; /* one-shot \crosstabview 1st argument */
+ char *crosstabview_col_H; /* one-shot \crosstabview 2nd argument */
bool notty; /* stdin or stdout is not a tty (as determined
* on startup) */
diff --git a/src/bin/psql/tab-complete.c b/src/bin/psql/tab-complete.c
index df678d5..bdaa92b 100644
--- a/src/bin/psql/tab-complete.c
+++ b/src/bin/psql/tab-complete.c
@@ -1172,7 +1172,8 @@ psql_completion(const char *text, int start, int end)
/* psql's backslash commands. */
static const char *const backslash_commands[] = {
- "\\a", "\\connect", "\\conninfo", "\\C", "\\cd", "\\copy", "\\copyright",
+ "\\a", "\\connect", "\\conninfo", "\\C", "\\cd", "\\copy",
+ "\\copyright", "\\crosstabview",
"\\d", "\\da", "\\db", "\\dc", "\\dC", "\\dd", "\\ddp", "\\dD",
"\\des", "\\det", "\\deu", "\\dew", "\\dE", "\\df",
"\\dF", "\\dFd", "\\dFp", "\\dFt", "\\dg", "\\di", "\\dl", "\\dL",
--
Sent via pgsql-hackers mailing list (pgsql-hackers@postgresql.org)
To make changes to your subscription:
http://www.postgresql.org/mailpref/pgsql-hackers
