feat: add S suffix for system-object variants of relation commands

changelog.rst

@@ -2,6 +2,10 @@
 Unreleased
 ==========
 
+Features:
+---------
+* Add the ``S`` modifier to relation commands so they can include system objects.
+
 Bug fixes:
 ----------
 * Include relation type/name titles in `\d` and `\d+` describe output so wildcard describe results retain per-relation context.

pgspecial/dbcommands.py

@@ -432,17 +432,28 @@ def _describe_extension(cur, oid):
         yield None, cur, headers, cur.statusmessage
 
 
-def list_objects(cur, pattern, verbose, relkinds):
+def list_objects(cur, pattern, verbose, relkinds, show_system=False):
     """
     Returns (title, rows, header, status)
 
     This method is used by list_tables, list_views, list_materialized views
     and list_indexes
 
     relkinds is a list of strings to filter pg_class.relkind
+    show_system includes objects from system schemas
 
     """
     schema_pattern, table_pattern = sql_name_pattern(pattern)
+    include_system_relations = show_system or bool(pattern)
+
+    # Match psql: system searches include legacy special relations and,
+    # for table listings, TOAST relations.
+    relkinds = list(relkinds)
+    if include_system_relations:
+        relkinds.append("s")
+
+        if "r" in relkinds:
+            relkinds.append("t")
 
     params = {"relkind": relkinds}
     if verbose:
@@ -460,9 +471,10 @@ def list_objects(cur, pattern, verbose, relkinds):
                     CASE c.relkind
                       WHEN 'r' THEN 'table' WHEN 'v' THEN 'view'
                       WHEN 'p' THEN 'partitioned table'
-                      WHEN 'm' THEN 'materialized view' WHEN 'i' THEN 'index'
+                      WHEN 'm' THEN 'materialized view'
+                      WHEN 'i' THEN 'index' WHEN 'I' THEN 'partitioned index'
                       WHEN 'S' THEN 'sequence' WHEN 's' THEN 'special'
-                      WHEN 'f' THEN 'foreign table' END
+                      WHEN 'f' THEN 'foreign table' WHEN 't' THEN 'TOAST table' END
                     as type,
                     pg_catalog.pg_get_userbyid(c.relowner) as owner
                     {verbose_columns}
@@ -478,6 +490,8 @@ def list_objects(cur, pattern, verbose, relkinds):
 
     if schema_pattern:
         params["schema_pattern"] = SQL(" AND n.nspname ~ {}").format(schema_pattern)
+    elif include_system_relations:
+        params["schema_pattern"] = SQL(" AND pg_catalog.pg_table_is_visible(c.oid) ")
     else:
         params["schema_pattern"] = SQL(
             """
@@ -506,24 +520,49 @@ def list_tables(cur, pattern, verbose):
     return list_objects(cur, pattern, verbose, ["r", "p", ""])
 
 
+@special_command("\\dtS", "\\dtS[+] [pattern]", "List user and system tables.")
+def list_tables_system(cur, pattern, verbose):
+    return list_objects(cur, pattern, verbose, ["r", "p", ""], show_system=True)
+
+
 @special_command("\\dv", "\\dv[+] [pattern]", "List views.")
 def list_views(cur, pattern, verbose):
-    return list_objects(cur, pattern, verbose, ["v", "s", ""])
+    return list_objects(cur, pattern, verbose, ["v", ""])
+
+
+@special_command("\\dvS", "\\dvS[+] [pattern]", "List user and system views.")
+def list_views_system(cur, pattern, verbose):
+    return list_objects(cur, pattern, verbose, ["v", ""], show_system=True)
 
 
 @special_command("\\dm", "\\dm[+] [pattern]", "List materialized views.")
 def list_materialized_views(cur, pattern, verbose):
-    return list_objects(cur, pattern, verbose, ["m", "s", ""])
+    return list_objects(cur, pattern, verbose, ["m", ""])
+
+
+@special_command("\\dmS", "\\dmS[+] [pattern]", "List user and system materialized views.")
+def list_materialized_views_system(cur, pattern, verbose):
+    return list_objects(cur, pattern, verbose, ["m", ""], show_system=True)
 
 
 @special_command("\\ds", "\\ds[+] [pattern]", "List sequences.")
 def list_sequences(cur, pattern, verbose):
-    return list_objects(cur, pattern, verbose, ["S", "s", ""])
+    return list_objects(cur, pattern, verbose, ["S", ""])
+
+
+@special_command("\\dsS", "\\dsS[+] [pattern]", "List user and system sequences.")
+def list_sequences_system(cur, pattern, verbose):
+    return list_objects(cur, pattern, verbose, ["S", ""], show_system=True)
 
 
 @special_command("\\di", "\\di[+] [pattern]", "List indexes.")
 def list_indexes(cur, pattern, verbose):
-    return list_objects(cur, pattern, verbose, ["i", "s", ""])
+    return list_objects(cur, pattern, verbose, ["i", "I", ""])
+
+
+@special_command("\\diS", "\\diS[+] [pattern]", "List user and system indexes.")
+def list_indexes_system(cur, pattern, verbose):
+    return list_objects(cur, pattern, verbose, ["i", "I", ""], show_system=True)
 
 
 @special_command("\\df", "\\df[+] [pattern]", "List functions.")
@@ -889,13 +928,22 @@ def _fetch_oid_details(cur, oid):
 @special_command("describe", "DESCRIBE [pattern]", "", hidden=True, case_sensitive=False)
 @special_command("\\d", "\\d[+] [pattern]", "List or describe tables, views and sequences.")
 def describe_table_details(cur, pattern, verbose):
+    return _describe_table_details(cur, pattern, verbose)
+
+
+@special_command("\\dS", "\\dS[+] [pattern]", "List or describe user and system relations.")
+def describe_table_details_system(cur, pattern, verbose):
+    return _describe_table_details(cur, pattern, verbose, show_system=True)
+
+
+def _describe_table_details(cur, pattern, verbose, show_system=False):
     """
     Returns (title, rows, headers, status)
     """
 
     # This is a simple \d[+] command. No table name to follow.
     if not pattern:
-        return list_objects(cur, pattern, verbose, ["r", "p", "v", "m", "S", "f", ""])
+        return list_objects(cur, pattern, verbose, ["r", "p", "v", "m", "S", "f", ""], show_system=show_system)
 
     # This is a \d <tablename> command. A royal pain in the ass.
     schema, relname = sql_name_pattern(pattern)
@@ -1925,51 +1973,14 @@ def shell_command(cur, pattern, verbose):
     return [(None, cur, headers, subprocess.call(params))]
 
 
-@special_command("\\dE", "\\dE[+] [pattern]", "List foreign tables.", aliases=())
+@special_command("\\dE", "\\dE[S+] [pattern]", "List foreign tables.", aliases=())
 def list_foreign_tables(cur, pattern, verbose):
-    params = {}
-    query = SQL(
-        """
-        SELECT n.nspname as schema,
-        c.relname as name,
-        CASE c.relkind WHEN 'r' THEN 'table' WHEN 'v' THEN 'view' WHEN 'm' THEN 'materialized view' WHEN 'i' THEN 'index' WHEN 'S' THEN 'sequence' WHEN 's' THEN 'special' WHEN 'f' THEN 'foreign table' WHEN 'p' THEN 'table' WHEN 'I' THEN 'index' END as type,
-        pg_catalog.pg_get_userbyid(c.relowner) as owner
-        {verbose_cols}
-        FROM pg_catalog.pg_class c
-            LEFT JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace
-        WHERE c.relkind IN ('f','')
-            AND n.nspname <> 'pg_catalog'
-            AND n.nspname <> 'information_schema'
-            AND n.nspname !~ '^pg_toast'
-        AND pg_catalog.pg_table_is_visible(c.oid)
-        {filter}
-        ORDER BY 1,2;
-        """
-    )
+    return list_objects(cur, pattern, verbose, ["f", ""])
 
-    if verbose:
-        params["verbose_cols"] = SQL(
-            """
-            , pg_catalog.pg_size_pretty(pg_catalog.pg_table_size(c.oid)) as size,
-            pg_catalog.obj_description(c.oid, 'pg_class') as description """
-        )
-    else:
-        params["verbose_cols"] = SQL("")
 
-    if pattern:
-        _, tbl_name = sql_name_pattern(pattern)
-        params["filter"] = SQL(" AND c.relname OPERATOR(pg_catalog.~) {} ").format(f"^({tbl_name})$")
-    else:
-        params["filter"] = SQL("")
-
-    formatted_query = query.format(**params)
-    log.debug(formatted_query.as_string(cur))
-    cur.execute(formatted_query)
-    if cur.description:
-        headers = [titleize(x.name) for x in cur.description]
-        return [(None, cur, headers, cur.statusmessage)]
-    else:
-        return [(None, None, None, cur.statusmessage)]
+@special_command("\\dES", "\\dES[+] [pattern]", "List user and system foreign tables.")
+def list_foreign_tables_system(cur, pattern, verbose):
+    return list_objects(cur, pattern, verbose, ["f", ""], show_system=True)
 
 
 def titleize(column):

tests/test_specials.py

@@ -136,6 +136,22 @@ def test_slash_d(executor):
     assert results == expected
 
 
+@dbtest
+def test_slash_d_system(executor):
+    rows = executor(r"\dS")[1]
+
+    assert ("public", "tbl1", "table", POSTGRES_USER) in rows
+    assert ("pg_catalog", "pg_class", "table", POSTGRES_USER) in rows
+    assert not any(row[1] == "pg_class_oid_index" for row in rows)
+
+
+@dbtest
+def test_slash_d_system_pattern(executor):
+    results = executor(r"\dS pg_class")
+
+    assert results[0] == 'Table "pg_catalog.pg_class"'
+
+
 @dbtest
 def test_slash_d_verbose(executor):
     results = executor(r"\d+")
@@ -520,6 +536,71 @@ def test_slash_dt(executor):
     assert results == expected
 
 
+@pytest.mark.parametrize(
+    ("command", "user_object", "system_object"),
+    [
+        (r"\dtS", ("public", "tbl1", "table", POSTGRES_USER), ("pg_catalog", "pg_class", "table", POSTGRES_USER)),
+        (r"\dvS", ("public", "vw1", "view", POSTGRES_USER), ("pg_catalog", "pg_views", "view", POSTGRES_USER)),
+        (r"\dmS", ("public", "mvw1", "materialized view", POSTGRES_USER), None),
+        (r"\dsS", ("public", "tbl2_id2_seq", "sequence", POSTGRES_USER), None),
+        (r"\diS", ("public", "id_text", "index", POSTGRES_USER), ("pg_catalog", "pg_class_oid_index", "index", POSTGRES_USER)),
+    ],
+)
+@dbtest
+def test_slash_d_relation_system_variants(executor, command, user_object, system_object):
+    rows = executor(command)[1]
+
+    assert user_object in rows
+    if system_object:
+        assert system_object in rows
+
+
+@dbtest
+@pytest.mark.skipif(SERVER_VERSION < 110000, reason="Partitioned indexes require PostgreSQL 11 or newer.")
+def test_slash_di_system_includes_partitioned_indexes(executor, connection):
+    with connection.cursor() as cursor:
+        cursor.execute("CREATE TABLE schema1.partitioned_tbl (id integer) PARTITION BY RANGE (id)")
+        cursor.execute("CREATE INDEX partitioned_idx ON schema1.partitioned_tbl (id)")
+        try:
+            rows = executor(r"\diS schema1.partitioned_idx")[1]
+        finally:
+            cursor.execute("DROP TABLE schema1.partitioned_tbl")
+
+    assert rows == [("schema1", "partitioned_idx", "partitioned index", POSTGRES_USER)]
+
+
+@dbtest
+def test_slash_dt_system_verbose(executor):
+    _, rows, headers, status = executor(r"\dtS+ tbl1")
+
+    assert headers == objects_listing_headers
+    assert len(rows) == 1
+    assert rows[0][:4] == ("public", "tbl1", "table", POSTGRES_USER)
+    assert status == "SELECT 1"
+
+
+@dbtest
+def test_slash_dt_pattern_includes_system_tables(executor):
+    rows = executor(r"\dt pg_class")[1]
+
+    assert rows == [("pg_catalog", "pg_class", "table", POSTGRES_USER)]
+
+
+@dbtest
+def test_slash_dt_system_respects_schema_visibility(executor):
+    rows = executor(r"\dtS")[1]
+
+    assert ("schema1", "s1_tbl1", "table", POSTGRES_USER) not in rows
+
+
+@dbtest
+def test_slash_dt_pattern_includes_toast_tables(executor):
+    rows = executor(r"\dt pg_toast.*")[1]
+
+    assert rows
+    assert all(schema == "pg_toast" and relation_type == "TOAST table" for schema, _, relation_type, _ in rows)
+
+
 @dbtest
 def test_slash_dt_verbose(executor):
     """List all tables in public schema in verbose mode."""
@@ -1137,6 +1218,32 @@ def test_slash_dE(executor):
         assert results == expected
 
 
+@fdw_test
+def test_slash_dE_system(executor):
+    with foreign_db_environ():
+        results = executor(r"\dES")
+        title = None
+        rows = [("public", "foreign_foo", "foreign table", "postgres")]
+        headers = ["Schema", "Name", "Type", "Owner"]
+        status = "SELECT 1"
+        expected = [title, rows, headers, status]
+        assert results == expected
+
+
+@fdw_test
+def test_slash_dE_system_with_schema_pattern(executor, connection):
+    with foreign_db_environ():
+        with connection.cursor() as cursor:
+            cursor.execute(
+                "CREATE FOREIGN TABLE schema1.foreign_foo (a int, b text) "
+                "SERVER foreign_db_server OPTIONS (schema_name 'public', table_name 'foreign_foo')"
+            )
+
+        results = executor(r"\dES schema1.foreign_foo")
+
+        assert results[1] == [("schema1", "foreign_foo", "foreign table", POSTGRES_USER)]
+
+
 @fdw_test
 def test_slash_dE_with_pattern(executor):
     with foreign_db_environ():