Add psql's \crosstabview

AUTHORS

@@ -156,6 +156,7 @@ Contributors:
     * Chris (ChrisJr404)
     * Pieter Ouwerkerk (pouwerkerk)
     * VXNCXNX
+    * Anand Hegde (anandghegde)
 
 Creator:
 --------

changelog.rst

@@ -1,7 +1,15 @@
 Upcoming
 ========
 
-
+Features:
+---------
+* Add psql's ``\crosstabview [colV [colH [colD [sortcolH]]]]``: run the query
+  and show the result as a crosstab grid, with the values of ``colV`` down the
+  side, the values of ``colH`` across the top and ``colD`` in the cells.
+  Columns are given by name or number and default to the first three, ``sortcolH``
+  orders the columns by an integer column, and a bare ``\crosstabview`` re-runs
+  the last query. The errors match psql's
+  ([issue 1378](https://github.com/dbcli/pgcli/issues/1378)).
 
 4.7.0 (2026-09-19)
 ==================

pgcli/crosstabview.py

@@ -0,0 +1,133 @@
+r"""psql's \crosstabview: show a query result as a crosstab (pivot) grid.
+
+The rules and error messages follow psql's crosstabview.c.
+"""
+
+import re
+
+# psql's CROSSTABVIEW_MAX_COLUMNS
+MAX_COLUMNS = 1600
+
+# "<query> \crosstabview [args]". An arg is a word or a "quoted" name; a single
+# quote in the tail means the \crosstabview is inside a string literal.
+_TERMINATOR = re.compile(r'^(.*?)\s*\\crosstabview((?:\s+(?:"[^"]*"|[^\s"\'])+)*)\s*$', re.DOTALL)
+
+
+class CrosstabViewError(Exception):
+    pass
+
+
+def split_query(sql):
+    r"""Split "<query> \crosstabview [args]" into (query, args), or None."""
+    match = _TERMINATOR.match(sql)
+    if match is None:
+        return None
+    return match.group(1), match.group(2)
+
+
+def parse_args(text):
+    """Split the arguments on whitespace, keeping "quoted" names whole."""
+    return re.findall(r'(?:"[^"]*"|[^\s"])+', text or "")
+
+
+def _dequote_downcase(arg):
+    """Like psql: downcase unquoted letters, strip quotes, "" means "."""
+    out = []
+    quoted = False
+    i = 0
+    while i < len(arg):
+        c = arg[i]
+        if c == '"':
+            if quoted and arg[i + 1 : i + 2] == '"':
+                out.append('"')
+                i += 1
+            else:
+                quoted = not quoted
+        else:
+            out.append(c if quoted or not "A" <= c <= "Z" else c.lower())
+        i += 1
+    return "".join(out)
+
+
+def _column_index(arg, headers):
+    """Resolve a column given by number (1-based) or by name."""
+    if arg.isdigit() and arg.isascii():
+        idx = int(arg) - 1
+        if not 0 <= idx < len(headers):
+            raise CrosstabViewError(f"\\crosstabview: column number {idx + 1} is out of range 1..{len(headers)}")
+        return idx
+    name = _dequote_downcase(arg)
+    matches = [i for i, header in enumerate(headers) if header == name]
+    if len(matches) > 1:
+        raise CrosstabViewError(f'\\crosstabview: ambiguous column name: "{name}"')
+    if not matches:
+        raise CrosstabViewError(f'\\crosstabview: column name not found: "{name}"')
+    return matches[0]
+
+
+def _key(value):
+    # psql compares the text of the values; NULL is a value of its own.
+    return None if value is None else str(value)
+
+
+def _rank(value):
+    # A sort value counts only when it is an integer; anything else ranks 0.
+    text = _key(value)
+    return int(text) if text is not None and re.fullmatch(r"-?[0-9]+", text) else 0
+
+
+def crosstabview(headers, rows, args):
+    """Pivot rows like psql's \\crosstabview colV colH [colD [sortcolH]].
+
+    Returns (rows, headers). A NULL horizontal header is returned as None;
+    cells with no data value are empty strings.
+    """
+    if len(headers) < 3:
+        raise CrosstabViewError("\\crosstabview: query must return at least three columns")
+    args = list(args[:4]) + [None] * (4 - len(args[:4]))
+
+    col_v = 0 if args[0] is None else _column_index(args[0], headers)
+    col_h = 1 if args[1] is None else _column_index(args[1], headers)
+    if col_v == col_h:
+        raise CrosstabViewError("\\crosstabview: vertical and horizontal headers must be different columns")
+    if args[2] is None:
+        if len(headers) != 3:
+            raise CrosstabViewError("\\crosstabview: data column must be specified when query returns more than three columns")
+        col_d = 3 - col_v - col_h  # the column that is left
+    else:
+        col_d = _column_index(args[2], headers)
+    col_sort = None if args[3] is None else _column_index(args[3], headers)
+
+    # Distinct header values, in order of first appearance.
+    h_ranks = {}
+    v_labels = {}
+    for row in rows:
+        h = _key(row[col_h])
+        if h not in h_ranks:
+            h_ranks[h] = 0 if col_sort is None else _rank(row[col_sort])
+            if len(h_ranks) > MAX_COLUMNS:
+                raise CrosstabViewError(f"\\crosstabview: maximum number of columns ({MAX_COLUMNS}) exceeded")
+        v_labels.setdefault(_key(row[col_v]), row[col_v])
+
+    h_order = list(h_ranks)
+    if col_sort is not None:
+        # Sort by rank; ties are in name order (NULL last), as in psql.
+        h_order.sort(key=lambda h: (h is None, h or ""))
+        h_order.sort(key=h_ranks.get)
+    h_index = {h: i for i, h in enumerate(h_order, 1)}
+    v_index = {v: i for i, v in enumerate(v_labels)}
+
+    empty = object()
+    grid = [[label] + [empty] * len(h_order) for label in v_labels.values()]
+    for row in rows:
+        v, h = _key(row[col_v]), _key(row[col_h])
+        cells = grid[v_index[v]]
+        if cells[h_index[h]] is not empty:
+            raise CrosstabViewError(
+                f'\\crosstabview: query result contains multiple data values for row "{"(null)" if v is None else v}", '
+                f'column "{"(null)" if h is None else h}"'
+            )
+        cells[h_index[h]] = row[col_d]
+
+    grid = [[("" if cell is empty else cell) for cell in cells] for cells in grid]
+    return grid, [headers[col_v]] + h_order

pgcli/main.py

@@ -500,13 +500,25 @@ def refresh_callback():
             "Echo a string to the query output channel.",
         )
 
+        self.pgspecial.register(
+            self.crosstabview,
+            "\\crosstabview",
+            "\\crosstabview [colV [colH [colD [sortcolH]]]]",
+            "Execute query (or re-run the last one) and display the result in a crosstab.",
+        )
+
         self.pgspecial.register(
             self.toggle_verbose_errors,
             "\\v",
             "\\v [on|off]",
             "Toggle verbose errors.",
         )
 
+    def crosstabview(self, pattern, **_):
+        # A bare \crosstabview re-runs the last query; "<query> \crosstabview"
+        # is handled by PGExecute.run().
+        return [self.pgexecute.crosstabview(None, pattern)]
+
     def toggle_verbose_errors(self, pattern, **_):
         flag = pattern.strip()
 
@@ -1296,6 +1308,10 @@ def _should_limit_output(self, sql, cur):
         if not is_select(sql):
             return False
 
+        # A \crosstabview grid is not a cursor; like psql, it is never truncated.
+        if not hasattr(cur, "rowcount"):
+            return False
+
         return not self._has_limit(sql) and self.row_limit != 0 and cur and cur.rowcount > self.row_limit
 
     def _has_limit(self, sql):
@@ -2194,7 +2210,8 @@ def format_status(cur, status):
         output.append(title)
 
     if cur:
-        headers = [] if settings.tuples_only else [case_function(x) for x in headers]
+        # A \crosstabview header can be NULL; show it like a NULL value.
+        headers = [] if settings.tuples_only else [settings.missingval if x is None else case_function(x) for x in headers]
         if max_width is not None:
             cur = list(cur)
         column_types = None

pgcli/pgbuffer.py

@@ -4,6 +4,7 @@
 from prompt_toolkit.enums import DEFAULT_BUFFER
 from prompt_toolkit.filters import Condition
 from prompt_toolkit.application import get_app
+from .crosstabview import split_query
 from .packages.parseutils.utils import is_open_quote
 
 _logger = logging.getLogger(__name__)
@@ -53,6 +54,7 @@ def cond():
             text.startswith("\\")
             or text.endswith((r"\e", r"\G"))
             or _is_complete(text)
+            or split_query(text) is not None
             or text == "exit"
             or text == "quit"
             or text == ":q"

pgcli/pgexecute.py

@@ -12,6 +12,7 @@
 sqlparse.engine.grouping.MAX_GROUPING_DEPTH = None
 sqlparse.engine.grouping.MAX_GROUPING_TOKENS = None
 
+from .crosstabview import CrosstabViewError, crosstabview, parse_args, split_query
 from .packages.parseutils.meta import FunctionMetadata, ForeignKey
 
 _logger = logging.getLogger(__name__)
@@ -182,6 +183,7 @@ def __init__(
         self.notify_callback = notify_callback
         self.connect(database, user, password, host, port, dsn, **kwargs)
         self.reset_expanded = None
+        self.last_sql = None
 
     def is_virtual_database(self):
         if self._is_virtual_database is None:
@@ -387,6 +389,15 @@ def run(
                             self.reset_expanded = True
                         sql = sql[:-2].strip()
 
+                    # "<query> \crosstabview [args]" runs the query and pivots it.
+                    crosstab = split_query(sql)
+                    if crosstab and crosstab[0]:
+                        if explain_mode:
+                            sql = crosstab[0]
+                        else:
+                            yield self.crosstabview(*crosstab) + (sql, True, False)
+                            continue
+
                     # First try to run each query as special
                     _logger.debug("Trying a pgspecial command. sql: %r", sql)
                     try:
@@ -418,6 +429,10 @@ def run(
                     sql = self.explain_prefix() + sql
 
                 yield self.execute_normal_sql(sql) + (sql, True, False)
+            except CrosstabViewError as e:
+                yield None, None, None, str(e), sql, False, False
+                if not on_error_resume:
+                    break
             except psycopg.DatabaseError as e:
                 _logger.error("sql: %r, error: %r", sql, e)
                 _logger.error("traceback: %r", traceback.format_exc())
@@ -451,6 +466,7 @@ def _must_raise(self, e):
     def execute_normal_sql(self, split_sql):
         """Returns tuple (title, rows, headers, status)"""
         _logger.debug("Regular sql statement. sql: %r", split_sql)
+        self.last_sql = split_sql
 
         title = ""
 
@@ -485,6 +501,25 @@ def handle_notices(n):
             _logger.debug("No rows in result.")
             return title, None, None, cur.statusmessage
 
+    def crosstabview(self, sql, args):
+        """Run sql (or the last query, like psql) and pivot the result.
+
+        Returns tuple (title, rows, headers, status)
+        """
+        sql = sql or self.last_sql
+        if not sql:
+            return None, None, None, "\\crosstabview: there is no previous query to run"
+        args = parse_args(args)
+        title, cur, headers, status = self.execute_normal_sql(sql)
+        if not headers:
+            # Not a result set: show it as usual, like psql does.
+            return title, cur, headers, status
+        rows, headers = crosstabview(headers, cur.fetchall(), args)
+        warnings = [f'\\crosstabview: extra argument "{arg}" ignored' for arg in args[4:]]
+        title = "\n".join(filter(None, [title] + warnings))
+        # An iterator, so an empty grid still prints its header like psql.
+        return title, iter(rows), headers, f"SELECT {len(rows)}"
+
     def search_path(self):
         """Returns the current search path as a list of schema names"""
 

tests/test_crosstabview.py

@@ -0,0 +1,83 @@
+import pytest
+
+from pgcli.crosstabview import CrosstabViewError, crosstabview, parse_args, split_query
+
+HEADERS = ["first", "second", "gt2"]
+ROWS = [(1, "one", False), (2, "two", False), (3, "three", True), (4, "four", True)]
+
+
+def test_split_query():
+    assert split_query("select 1") is None
+    assert split_query("select 1 \\crosstabview") == ("select 1", "")
+    assert split_query("select 1\n\\crosstabview a \"B c\"") == ("select 1", ' a "B c"')
+    assert split_query("\\crosstabview 1 2") == ("", " 1 2")
+    # inside a string literal
+    assert split_query("select 'x \\crosstabview y'") is None
+
+
+def test_parse_args():
+    assert parse_args(' a "B c"  3 "fO""o"') == ["a", '"B c"', "3", '"fO""o"']
+    assert parse_args("") == []
+
+
+def test_defaults():
+    # first column vertical, second horizontal, third data
+    rows, headers = crosstabview(HEADERS, ROWS, [])
+    assert headers == ["first", "one", "two", "three", "four"]
+    assert rows == [
+        [1, False, "", "", ""],
+        [2, "", False, "", ""],
+        [3, "", "", True, ""],
+        [4, "", "", "", True],
+    ]
+
+
+def test_columns_by_name_and_number():
+    rows, headers = crosstabview(HEADERS, ROWS, ["second", "1", "GT2"])
+    assert headers == ["second", "1", "2", "3", "4"]
+    assert rows[0] == ["one", False, "", "", ""]
+
+
+def test_quoted_names_are_case_sensitive():
+    headers = ["Foo", "b", "c"]
+    assert crosstabview(headers, [(1, 2, 3)], ['"Foo"'])[1] == ["Foo", "2"]
+    with pytest.raises(CrosstabViewError, match=r'column name not found: "foo"'):
+        crosstabview(headers, [(1, 2, 3)], ["Foo"])
+    # a quoted number is a name, and "" is an escaped quote
+    assert crosstabview(["1", 'fO"o', "c"], [(1, 2, 3)], ['"fO""o"', '"1"'])[1] == ['fO"o', "1"]
+
+
+def test_sort_column():
+    headers = ["a", "b", "c", "s"]
+    rows = [("x", "b", 1, "z"), ("x", "a", 2, "-3"), ("y", "c", 3, None), ("y", "d", 3, "1.5"), ("y", "e", 4, "-1")]
+    grid, headers = crosstabview(headers, rows, ["a", "b", "c", "s"])
+    # integer ranks sort; anything else ranks 0, ties in name order
+    assert headers == ["a", "a", "e", "b", "c", "d"]
+    assert grid == [["x", 2, "", 1, "", ""], ["y", "", 4, "", 3, 3]]
+
+
+def test_nulls():
+    rows = [(1, 2, None), (1, None, 4), (None, 3, 5), (2, 3, 6)]
+    grid, headers = crosstabview(["a", "b", "c"], rows, [])
+    # A NULL data value stays None; a cell without data is empty.
+    assert headers == ["a", "2", None, "3"]
+    assert grid == [[1, None, 4, ""], [None, "", "", 5], [2, "", "", 6]]
+
+
+@pytest.mark.parametrize(
+    "headers, rows, args, message",
+    [
+        (["a", "b"], [], [], "query must return at least three columns"),
+        (["a", "b", "c", "d"], [], [], "data column must be specified when query returns more than three columns"),
+        (["a", "b", "c"], [], ["a", "a"], "vertical and horizontal headers must be different columns"),
+        (["a", "b", "c"], [], ["5"], r"column number 5 is out of range 1\.\.3"),
+        (["a", "b", "c"], [], ["0"], r"column number 0 is out of range 1\.\.3"),
+        (["a", "a", "c"], [], ["a"], 'ambiguous column name: "a"'),
+        (["a", "b", "c"], [(1, 2, 3), (1, 2, 4)], [], 'multiple data values for row "1", column "2"'),
+        (["a", "b", "c"], [(1, None, 3), (1, None, 4)], [], r'multiple data values for row "1", column "\(null\)"'),
+        (["a", "b", "c"], [(1, i, 1) for i in range(1601)], [], r"maximum number of columns \(1600\) exceeded"),
+    ],
+)
+def test_errors(headers, rows, args, message):
+    with pytest.raises(CrosstabViewError, match=message):
+        crosstabview(headers, rows, args)