Show shared object statistics in pg_stat_database

This adds a row to the pg_stat_database view with datoid 0 and datname
NULL for those objects that are not in a database. This was added
particularly for checksums, but we were already tracking more satistics
for these objects, just not returning it.

Also add a checksum_last_failure column that holds the timestamptz of
the last checksum failure that occurred in a database (or in a
non-dataabase file), if any.

Author: Julien Rouhaud <rjuju123@gmail.com>

doc/src/sgml/monitoring.sgml

@@ -2498,20 +2498,22 @@ SELECT pid, wait_event_type, wait_event FROM pg_stat_activity WHERE wait_event i
     <row>
      <entry><structfield>datid</structfield></entry>
      <entry><type>oid</type></entry>
-     <entry>OID of a database</entry>
+     <entry>OID of this database, or 0 for objects belonging to a shared
+     relation</entry>
     </row>
     <row>
      <entry><structfield>datname</structfield></entry>
      <entry><type>name</type></entry>
-     <entry>Name of this database</entry>
+     <entry>Name of this database, or <literal>NULL</literal> for the shared
+     objects.</entry>
     </row>
     <row>
      <entry><structfield>numbackends</structfield></entry>
      <entry><type>integer</type></entry>
-     <entry>Number of backends currently connected to this database.
+     <entry>Number of backends currently connected to this database, or
-     This is the only column in this view that returns a value reflecting
+     <literal>NULL</literal> for the shared objects.  This is the only column
-     current state; all other columns return the accumulated values since
+     in this view that returns a value reflecting current state; all other
-     the last reset.</entry>
+     columns return the accumulated values since the last reset.</entry>
     </row>
     <row>
      <entry><structfield>xact_commit</structfield></entry>
@@ -2597,7 +2599,14 @@ SELECT pid, wait_event_type, wait_event FROM pg_stat_activity WHERE wait_event i
     <row>
      <entry><structfield>checksum_failures</structfield></entry>
      <entry><type>bigint</type></entry>
-     <entry>Number of data page checksum failures detected in this database</entry>
+     <entry>Number of data page checksum failures detected in this
+     database</entry>
+    </row>
+    <row>
+     <entry><structfield>checksum_last_failure</structfield></entry>
+     <entry><type>timestamp with time zone</type></entry>
+     <entry>Time at which the last data page checksum failure was detected in
+     this database, or on a shared object.</entry>
     </row>
     <row>
      <entry><structfield>blk_read_time</structfield></entry>
@@ -2622,7 +2631,8 @@ SELECT pid, wait_event_type, wait_event FROM pg_stat_activity WHERE wait_event i
 
   <para>
    The <structname>pg_stat_database</structname> view will contain one row
-   for each database in the cluster, showing database-wide statistics.
+   for each database in the cluster, plus one for the shared objects, showing
+   database-wide statistics.
   </para>
 
   <table id="pg-stat-database-conflicts-view" xreflabel="pg_stat_database_conflicts">

doc/src/sgml/ref/initdb.sgml

@@ -218,7 +218,9 @@ PostgreSQL documentation
         I/O system that would otherwise be silent. Enabling checksums
         may incur a noticeable performance penalty. This option can only
         be set during initialization, and cannot be changed later. If
-        set, checksums are calculated for all objects, in all databases.
+        set, checksums are calculated for all objects, in all databases. All
+        checksum failures will be reported in the <xref
+        linkend="pg-stat-database-view"/> view.
        </para>
       </listitem>
      </varlistentry>

doc/src/sgml/ref/pg_basebackup.sgml

@@ -531,7 +531,8 @@ PostgreSQL documentation
         By default, checksums are verified and checksum failures will result
         in a non-zero exit status. However, the base backup will not be
         removed in such a case, as if the <option>--no-clean</option> option
-        had been used.
+        had been used.  Checksum verifications failures will also be reported
+        in the <xref linkend="pg-stat-database-view"/> view.
        </para>
       </listitem>
      </varlistentry>

src/backend/catalog/system_views.sql

@@ -816,7 +816,10 @@ CREATE VIEW pg_stat_database AS
     SELECT
             D.oid AS datid,
             D.datname AS datname,
-            pg_stat_get_db_numbackends(D.oid) AS numbackends,
+                CASE
+                    WHEN (D.oid = (0)::oid) THEN NULL::integer
+                    ELSE pg_stat_get_db_numbackends(D.oid)
+                END AS numbackends,
             pg_stat_get_db_xact_commit(D.oid) AS xact_commit,
             pg_stat_get_db_xact_rollback(D.oid) AS xact_rollback,
             pg_stat_get_db_blocks_fetched(D.oid) -
@@ -832,10 +835,15 @@ CREATE VIEW pg_stat_database AS
             pg_stat_get_db_temp_bytes(D.oid) AS temp_bytes,
             pg_stat_get_db_deadlocks(D.oid) AS deadlocks,
             pg_stat_get_db_checksum_failures(D.oid) AS checksum_failures,
+            pg_stat_get_db_checksum_last_failure(D.oid) AS checksum_last_failure,
             pg_stat_get_db_blk_read_time(D.oid) AS blk_read_time,
             pg_stat_get_db_blk_write_time(D.oid) AS blk_write_time,
             pg_stat_get_db_stat_reset_time(D.oid) AS stats_reset
-    FROM pg_database D;
+    FROM (
+        SELECT 0 AS oid, NULL::name AS datname
+        UNION ALL
+        SELECT oid, datname FROM pg_database
+    ) D;
 
 CREATE VIEW pg_stat_database_conflicts AS
     SELECT

src/backend/postmaster/pgstat.c

@@ -1523,7 +1523,7 @@ pgstat_report_deadlock(void)
 
 
 /* --------
- * pgstat_report_checksum_failures_in_db(dboid, failure_count) -
+ * pgstat_report_checksum_failures_in_db() -
  *
  *	Tell the collector about one or more checksum failures.
  * --------
@@ -1539,6 +1539,8 @@ pgstat_report_checksum_failures_in_db(Oid dboid, int failurecount)
 	pgstat_setheader(&msg.m_hdr, PGSTAT_MTYPE_CHECKSUMFAILURE);
 	msg.m_databaseid = dboid;
 	msg.m_failurecount = failurecount;
+	msg.m_failure_time = GetCurrentTimestamp();
+
 	pgstat_send(&msg, sizeof(msg));
 }
 
@@ -4651,6 +4653,7 @@ reset_dbentry_counters(PgStat_StatDBEntry *dbentry)
 	dbentry->n_temp_bytes = 0;
 	dbentry->n_deadlocks = 0;
 	dbentry->n_checksum_failures = 0;
+	dbentry->last_checksum_failure = 0;
 	dbentry->n_block_read_time = 0;
 	dbentry->n_block_write_time = 0;
 
@@ -6307,6 +6310,7 @@ pgstat_recv_checksum_failure(PgStat_MsgChecksumFailure *msg, int len)
 	dbentry = pgstat_get_db_entry(msg->m_databaseid, true);
 
 	dbentry->n_checksum_failures += msg->m_failurecount;
+	dbentry->last_checksum_failure = msg->m_failure_time;
 }
 
 /* ----------

src/backend/replication/basebackup.c

@@ -1584,9 +1584,9 @@ sendFile(const char *readfilename, const char *tarfilename, struct stat *statbuf
 				(errmsg("file \"%s\" has a total of %d checksum verification "
 						"failures", readfilename, checksum_failures)));
 
-		if (dboid != InvalidOid)
 		pgstat_report_checksum_failures_in_db(dboid, checksum_failures);
 	}
+
 	total_checksum_failures += checksum_failures;
 
 	return true;

src/backend/utils/adt/pgstatfuncs.c

@@ -1534,6 +1534,24 @@ pg_stat_get_db_checksum_failures(PG_FUNCTION_ARGS)
 	PG_RETURN_INT64(result);
 }
 
+Datum
+pg_stat_get_db_checksum_last_failure(PG_FUNCTION_ARGS)
+{
+	Oid			dbid = PG_GETARG_OID(0);
+	TimestampTz result;
+	PgStat_StatDBEntry *dbentry;
+
+	if ((dbentry = pgstat_fetch_stat_dbentry(dbid)) == NULL)
+		result = 0;
+	else
+		result = dbentry->last_checksum_failure;
+
+	if (result == 0)
+		PG_RETURN_NULL();
+	else
+		PG_RETURN_TIMESTAMPTZ(result);
+}
+
 Datum
 pg_stat_get_db_blk_read_time(PG_FUNCTION_ARGS)
 {

src/include/catalog/pg_proc.dat

@@ -5285,6 +5285,11 @@
   proname => 'pg_stat_get_db_checksum_failures', provolatile => 's',
   proparallel => 'r', prorettype => 'int8', proargtypes => 'oid',
   prosrc => 'pg_stat_get_db_checksum_failures' },
+{ oid => '8394',
+  descr => 'statistics: when last checksum failure was detected in database',
+  proname => 'pg_stat_get_db_checksum_last_failure', provolatile => 's',
+  proparallel => 'r', prorettype => 'timestamptz', proargtypes => 'oid',
+  prosrc => 'pg_stat_get_db_checksum_last_failure' },
 { oid => '3074', descr => 'statistics: last reset for a database',
   proname => 'pg_stat_get_db_stat_reset_time', provolatile => 's',
   proparallel => 'r', prorettype => 'timestamptz', proargtypes => 'oid',

src/include/pgstat.h

@@ -541,6 +541,7 @@ typedef struct PgStat_MsgChecksumFailure
 	PgStat_MsgHdr m_hdr;
 	Oid			m_databaseid;
 	int			m_failurecount;
+	TimestampTz	m_failure_time;
 } PgStat_MsgChecksumFailure;
 
 
@@ -607,6 +608,7 @@ typedef struct PgStat_StatDBEntry
 	PgStat_Counter n_temp_bytes;
 	PgStat_Counter n_deadlocks;
 	PgStat_Counter n_checksum_failures;
+	TimestampTz last_checksum_failure;
 	PgStat_Counter n_block_read_time;	/* times in microseconds */
 	PgStat_Counter n_block_write_time;
 

src/test/regress/expected/rules.out

@@ -1805,7 +1805,10 @@ pg_stat_bgwriter| SELECT pg_stat_get_bgwriter_timed_checkpoints() AS checkpoints
     pg_stat_get_bgwriter_stat_reset_time() AS stats_reset;
 pg_stat_database| SELECT d.oid AS datid,
     d.datname,
-    pg_stat_get_db_numbackends(d.oid) AS numbackends,
+        CASE
+            WHEN (d.oid = (0)::oid) THEN NULL::integer
+            ELSE pg_stat_get_db_numbackends(d.oid)
+        END AS numbackends,
     pg_stat_get_db_xact_commit(d.oid) AS xact_commit,
     pg_stat_get_db_xact_rollback(d.oid) AS xact_rollback,
     (pg_stat_get_db_blocks_fetched(d.oid) - pg_stat_get_db_blocks_hit(d.oid)) AS blks_read,
@@ -1820,10 +1823,16 @@ pg_stat_database| SELECT d.oid AS datid,
     pg_stat_get_db_temp_bytes(d.oid) AS temp_bytes,
     pg_stat_get_db_deadlocks(d.oid) AS deadlocks,
     pg_stat_get_db_checksum_failures(d.oid) AS checksum_failures,
+    pg_stat_get_db_checksum_last_failure(d.oid) AS checksum_last_failure,
     pg_stat_get_db_blk_read_time(d.oid) AS blk_read_time,
     pg_stat_get_db_blk_write_time(d.oid) AS blk_write_time,
     pg_stat_get_db_stat_reset_time(d.oid) AS stats_reset
-   FROM pg_database d;
+   FROM ( SELECT 0 AS oid,
+            NULL::name AS datname
+        UNION ALL
+         SELECT pg_database.oid,
+            pg_database.datname
+           FROM pg_database) d;
 pg_stat_database_conflicts| SELECT d.oid AS datid,
     d.datname,
     pg_stat_get_db_conflict_tablespace(d.oid) AS confl_tablespace,