crosstabview.c File Reference

#include "postgres_fe.h"
#include "common.h"
#include "common/int.h"
#include "common/logging.h"
#include "crosstabview.h"
#include "pqexpbuffer.h"
#include "psqlscanslash.h"
#include "settings.h"

Include dependency graph for crosstabview.c:

Go to the source code of this file.

Data Structures
struct	_pivot_field
struct	_avl_node
struct	_avl_tree

Typedefs
typedef struct _pivot_field	pivot_field
typedef struct _avl_node	avl_node
typedef struct _avl_tree	avl_tree

Functions
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)
bool	PrintResultInCrosstab (const PGresult *res)
static void	avlUpdateHeight (avl_node *n)
static avl_node *	avlRotate (avl_node **current, int dir)
static int	avlBalance (avl_node *n)
static void	avlAdjustBalance (avl_tree *tree, avl_node **node)
static void	avlInsertNode (avl_tree *tree, avl_node **node, pivot_field field)

Typedef Documentation

avl_node

typedef struct _avl_node avl_node

avl_tree

typedef struct _avl_tree avl_tree

pivot_field

typedef struct _pivot_field pivot_field

Function Documentation

avlAdjustBalance()

static void avlAdjustBalance ( avl_tree *  tree, avl_node **  node  )

static

Definition at line 504 of file crosstabview.c.

{
    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);
}

References avlBalance(), avlRotate(), avlUpdateHeight(), b, _avl_node::children, and tree.

Referenced by avlInsertNode().

avlBalance()

static int avlBalance ( avl_node *  n )

static

Definition at line 493 of file crosstabview.c.

{
    return n->children[0]->height - n->children[1]->height;
}

References _avl_node::children, and _avl_node::height.

Referenced by avlAdjustBalance().

avlCollectFields()

static int avlCollectFields ( avl_tree *  tree, avl_node *  node, pivot_field *  fields, int  idx  )

static

Definition at line 574 of file crosstabview.c.

{
    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);
}

References avlCollectFields(), _avl_node::children, _avl_node::field, idx(), and tree.

Referenced by avlCollectFields(), and PrintResultInCrosstab().

avlFree()

static void avlFree ( avl_tree *  tree, avl_node *  node  )

static

Definition at line 446 of file crosstabview.c.

{
    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);
    }
}

References avlFree(), _avl_node::children, pg_free(), and tree.

Referenced by avlFree(), and PrintResultInCrosstab().

avlInit()

static void avlInit ( avl_tree *  tree )

static

Definition at line 436 of file crosstabview.c.

{
    tree->end = pg_malloc0_object(avl_node);
    tree->end->children[0] = tree->end->children[1] = tree->end;
    tree->count = 0;
    tree->root = tree->end;
}

References pg_malloc0_object, and tree.

Referenced by PrintResultInCrosstab().

avlInsertNode()

static void avlInsertNode ( avl_tree *  tree, avl_node **  node, pivot_field  field  )

static

Definition at line 527 of file crosstabview.c.

{
    avl_node   *current = *node;
 
    if (current == tree->end)
    {
        avl_node   *new_node = pg_malloc_object(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);
        }
    }
}

References avlAdjustBalance(), avlInsertNode(), _avl_node::children, cmp(), fb(), _avl_node::field, pg_malloc_object, pivotFieldCompare(), and tree.

Referenced by avlInsertNode(), and avlMergeValue().

avlMergeValue()

static void avlMergeValue ( avl_tree *  tree, char *  name, char *  sort_value  )

static

Definition at line 557 of file crosstabview.c.

{
    pivot_field field;
 
    field.name = name;
    field.rank = tree->count;
    field.sort_value = sort_value;
    avlInsertNode(tree, &tree->root, field);
}

References avlInsertNode(), name, _pivot_field::name, _pivot_field::rank, _pivot_field::sort_value, and tree.

Referenced by PrintResultInCrosstab().

avlRotate()

static avl_node * avlRotate ( avl_node **  current, int  dir  )

static

Definition at line 479 of file crosstabview.c.

{
    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;
}

References avlUpdateHeight(), before(), _avl_node::children, and fb().

Referenced by avlAdjustBalance().

avlUpdateHeight()

static void avlUpdateHeight ( avl_node *  n )

static

Definition at line 470 of file crosstabview.c.

{
    n->height = 1 + (n->children[0]->height > n->children[1]->height ?
                     n->children[0]->height :
                     n->children[1]->height);
}

References _avl_node::children, and _avl_node::height.

Referenced by avlAdjustBalance(), and avlRotate().

indexOfColumn()

static int indexOfColumn ( char *  arg, const PGresult *  res  )

static

Definition at line 633 of file crosstabview.c.

{
    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;
}

References arg, dequote_downcase_identifier(), _psqlSettings::encoding, fb(), i, idx(), pg_log_error, PQfname, PQnfields, and pset.

Referenced by PrintResultInCrosstab().

pivotFieldCompare()

static int pivotFieldCompare ( const void *  a, const void *  b  )

static

Definition at line 692 of file crosstabview.c.

{
    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);
}

References a, b, fb(), and _pivot_field::name.

Referenced by avlInsertNode(), and printCrosstab().

printCrosstab()

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

Definition at line 284 of file crosstabview.c.

{
    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 = pg_malloc_array(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;
}

References Assert, column_type_alignment(), error(), fb(), i, idx(), _psqlSettings::logfile, name, _pivot_field::name, printQueryOpt::nullPrint, pg_free(), pg_log_error, pg_malloc_array, pivotFieldCompare(), _psqlSettings::popt, PQfname, PQftype(), PQgetisnull, PQgetvalue, PQntuples, printTable(), printTableAddHeader(), printTableCleanup(), printTableInit(), pset, _psqlSettings::queryFout, printQueryOpt::title, and printQueryOpt::topt.

Referenced by PrintResultInCrosstab().

PrintResultInCrosstab()

bool PrintResultInCrosstab

(

const PGresult *

res

)

Definition at line 104 of file crosstabview.c.

{
    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 = pg_malloc_array(pivot_field, num_columns);
 
    array_rows = pg_malloc_array(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;
}

References Assert, avlCollectFields(), avlFree(), avlInit(), avlMergeValue(), CROSSTABVIEW_MAX_COLUMNS, _psqlSettings::ctv_args, fb(), i, indexOfColumn(), pg_free(), pg_log_error, pg_malloc_array, PGRES_TUPLES_OK, PQgetisnull, PQgetvalue, PQnfields, PQntuples, PQresultStatus, printCrosstab(), pset, rankSort(), and val.

Referenced by PrintQueryResult().

rankCompare()

static int rankCompare ( const void *  a, const void *  b  )

static

Definition at line 708 of file crosstabview.c.

{
    return pg_cmp_s32(*(const int *) a, *(const int *) b);
}

References a, b, and pg_cmp_s32().

Referenced by rankSort().

rankSort()

static void rankSort ( int  num_columns, pivot_field *  piv_columns  )

static

Definition at line 585 of file crosstabview.c.

{
    int        *hmap;           /* [[offset in piv_columns, rank], ...for
                                 * every header entry] */
    int         i;
 
    hmap = pg_malloc_array(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);
}

References fb(), i, pg_free(), pg_malloc_array, qsort, rankCompare(), and val.

Referenced by PrintResultInCrosstab().