/src/postgres/src/backend/utils/adt/pgstatfuncs.c

Source (jump to first uncovered line)
/*-------------------------------------------------------------------------
 *
 * pgstatfuncs.c
 *    Functions for accessing various forms of statistics data
 *
 * Portions Copyright (c) 1996-2025, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, Regents of the University of California
 *
 *
 * IDENTIFICATION
 *    src/backend/utils/adt/pgstatfuncs.c
 *
 *-------------------------------------------------------------------------
 */
#include "postgres.h"

#include "access/htup_details.h"
#include "access/xlog.h"
#include "access/xlogprefetcher.h"
#include "catalog/catalog.h"
#include "catalog/pg_authid.h"
#include "catalog/pg_type.h"
#include "common/ip.h"
#include "funcapi.h"
#include "miscadmin.h"
#include "pgstat.h"
#include "postmaster/bgworker.h"
#include "replication/logicallauncher.h"
#include "storage/proc.h"
#include "storage/procarray.h"
#include "utils/acl.h"
#include "utils/builtins.h"
#include "utils/timestamp.h"

#define UINT32_ACCESS_ONCE(var)    ((uint32)(*((volatile uint32 *)&(var))))

#define HAS_PGSTAT_PERMISSIONS(role)   (has_privs_of_role(GetUserId(), ROLE_PG_READ_ALL_STATS) || has_privs_of_role(GetUserId(), role))

#define PG_STAT_GET_RELENTRY_INT64(stat)            \
Datum                             \
CppConcat(pg_stat_get_,stat)(PG_FUNCTION_ARGS)          \
{                               \
  Oid     relid = PG_GETARG_OID(0);            \
  int64   result;                     \
  PgStat_StatTabEntry *tabentry;                \
                                \
  if ((tabentry = pgstat_fetch_stat_tabentry(relid)) == NULL) \
    result = 0;                       \
  else                            \
    result = (int64) (tabentry->stat);           \
                                \
  PG_RETURN_INT64(result);                 \
}

/* pg_stat_get_analyze_count */
PG_STAT_GET_RELENTRY_INT64(analyze_count)

/* pg_stat_get_autoanalyze_count */
PG_STAT_GET_RELENTRY_INT64(autoanalyze_count)

/* pg_stat_get_autovacuum_count */
PG_STAT_GET_RELENTRY_INT64(autovacuum_count)

/* pg_stat_get_blocks_fetched */
PG_STAT_GET_RELENTRY_INT64(blocks_fetched)

/* pg_stat_get_blocks_hit */
PG_STAT_GET_RELENTRY_INT64(blocks_hit)

/* pg_stat_get_dead_tuples */
PG_STAT_GET_RELENTRY_INT64(dead_tuples)

/* pg_stat_get_ins_since_vacuum */
PG_STAT_GET_RELENTRY_INT64(ins_since_vacuum)

/* pg_stat_get_live_tuples */
PG_STAT_GET_RELENTRY_INT64(live_tuples)

/* pg_stat_get_mod_since_analyze */
PG_STAT_GET_RELENTRY_INT64(mod_since_analyze)

/* pg_stat_get_numscans */
PG_STAT_GET_RELENTRY_INT64(numscans)

/* pg_stat_get_tuples_deleted */
PG_STAT_GET_RELENTRY_INT64(tuples_deleted)

/* pg_stat_get_tuples_fetched */
PG_STAT_GET_RELENTRY_INT64(tuples_fetched)

/* pg_stat_get_tuples_hot_updated */
PG_STAT_GET_RELENTRY_INT64(tuples_hot_updated)

/* pg_stat_get_tuples_newpage_updated */
PG_STAT_GET_RELENTRY_INT64(tuples_newpage_updated)

/* pg_stat_get_tuples_inserted */
PG_STAT_GET_RELENTRY_INT64(tuples_inserted)

/* pg_stat_get_tuples_returned */
PG_STAT_GET_RELENTRY_INT64(tuples_returned)

/* pg_stat_get_tuples_updated */
PG_STAT_GET_RELENTRY_INT64(tuples_updated)

/* pg_stat_get_vacuum_count */
PG_STAT_GET_RELENTRY_INT64(vacuum_count)

#define PG_STAT_GET_RELENTRY_FLOAT8(stat)           \
Datum                             \
CppConcat(pg_stat_get_,stat)(PG_FUNCTION_ARGS)          \
{                               \
  Oid     relid = PG_GETARG_OID(0);            \
  double    result;                     \
  PgStat_StatTabEntry *tabentry;                \
                                \
  if ((tabentry = pgstat_fetch_stat_tabentry(relid)) == NULL) \
    result = 0;                       \
  else                            \
    result = (double) (tabentry->stat);           \
                                \
  PG_RETURN_FLOAT8(result);                 \
}

/* pg_stat_get_total_vacuum_time */
PG_STAT_GET_RELENTRY_FLOAT8(total_vacuum_time)

/* pg_stat_get_total_autovacuum_time */
PG_STAT_GET_RELENTRY_FLOAT8(total_autovacuum_time)

/* pg_stat_get_total_analyze_time */
PG_STAT_GET_RELENTRY_FLOAT8(total_analyze_time)

/* pg_stat_get_total_autoanalyze_time */
PG_STAT_GET_RELENTRY_FLOAT8(total_autoanalyze_time)

#define PG_STAT_GET_RELENTRY_TIMESTAMPTZ(stat)          \
Datum                             \
CppConcat(pg_stat_get_,stat)(PG_FUNCTION_ARGS)          \
{                               \
  Oid     relid = PG_GETARG_OID(0);            \
  TimestampTz result;                     \
  PgStat_StatTabEntry *tabentry;                \
                                \
  if ((tabentry = pgstat_fetch_stat_tabentry(relid)) == NULL) \
    result = 0;                       \
  else                            \
    result = tabentry->stat;               \
                                \
  if (result == 0)                     \
    PG_RETURN_NULL();                   \
  else                            \
    PG_RETURN_TIMESTAMPTZ(result);             \
}

/* pg_stat_get_last_analyze_time */
PG_STAT_GET_RELENTRY_TIMESTAMPTZ(last_analyze_time)

/* pg_stat_get_last_autoanalyze_time */
PG_STAT_GET_RELENTRY_TIMESTAMPTZ(last_autoanalyze_time)

/* pg_stat_get_last_autovacuum_time */
PG_STAT_GET_RELENTRY_TIMESTAMPTZ(last_autovacuum_time)

/* pg_stat_get_last_vacuum_time */
PG_STAT_GET_RELENTRY_TIMESTAMPTZ(last_vacuum_time)

/* pg_stat_get_lastscan */
PG_STAT_GET_RELENTRY_TIMESTAMPTZ(lastscan)

Datum
pg_stat_get_function_calls(PG_FUNCTION_ARGS)
{
  Oid     funcid = PG_GETARG_OID(0);
  PgStat_StatFuncEntry *funcentry;

  if ((funcentry = pgstat_fetch_stat_funcentry(funcid)) == NULL)
    PG_RETURN_NULL();
  PG_RETURN_INT64(funcentry->numcalls);
}

/* convert counter from microsec to millisec for display */
#define PG_STAT_GET_FUNCENTRY_FLOAT8_MS(stat)           \
Datum                               \
CppConcat(pg_stat_get_function_,stat)(PG_FUNCTION_ARGS)       \
{                                 \
  Oid     funcid = PG_GETARG_OID(0);              \
  double    result;                       \
  PgStat_StatFuncEntry *funcentry;                \
                                  \
  if ((funcentry = pgstat_fetch_stat_funcentry(funcid)) == NULL) \
    PG_RETURN_NULL();                     \
  result = ((double) funcentry->stat) / 1000.0;         \
  PG_RETURN_FLOAT8(result);                   \
}

/* pg_stat_get_function_total_time */
PG_STAT_GET_FUNCENTRY_FLOAT8_MS(total_time)

/* pg_stat_get_function_self_time */
PG_STAT_GET_FUNCENTRY_FLOAT8_MS(self_time)

Datum
pg_stat_get_backend_idset(PG_FUNCTION_ARGS)
{
  FuncCallContext *funcctx;
  int      *fctx;

  /* stuff done only on the first call of the function */
  if (SRF_IS_FIRSTCALL())
  {
    /* create a function context for cross-call persistence */
    funcctx = SRF_FIRSTCALL_INIT();

    fctx = MemoryContextAlloc(funcctx->multi_call_memory_ctx,
                  sizeof(int));
    funcctx->user_fctx = fctx;

    fctx[0] = 0;
  }

  /* stuff done on every call of the function */
  funcctx = SRF_PERCALL_SETUP();
  fctx = funcctx->user_fctx;

  fctx[0] += 1;

  /*
   * We recheck pgstat_fetch_stat_numbackends() each time through, just in
   * case the local status data has been refreshed since we started.  It's
   * plenty cheap enough if not.  If a refresh does happen, we'll likely
   * miss or duplicate some backend IDs, but we're content not to crash.
   * (Refreshing midway through such a query would be problematic usage
   * anyway, since the backend IDs we've already returned might no longer
   * refer to extant sessions.)
   */
  if (fctx[0] <= pgstat_fetch_stat_numbackends())
  {
    /* do when there is more left to send */
    LocalPgBackendStatus *local_beentry = pgstat_get_local_beentry_by_index(fctx[0]);

    SRF_RETURN_NEXT(funcctx, Int32GetDatum(local_beentry->proc_number));
  }
  else
  {
    /* do when there is no more left */
    SRF_RETURN_DONE(funcctx);
  }
}

/*
 * Returns command progress information for the named command.
 */
Datum
pg_stat_get_progress_info(PG_FUNCTION_ARGS)
{
#define PG_STAT_GET_PROGRESS_COLS PGSTAT_NUM_PROGRESS_PARAM + 3
  int     num_backends = pgstat_fetch_stat_numbackends();
  int     curr_backend;
  char     *cmd = text_to_cstring(PG_GETARG_TEXT_PP(0));
  ProgressCommandType cmdtype;
  ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;

  /* Translate command name into command type code. */
  if (pg_strcasecmp(cmd, "VACUUM") == 0)
    cmdtype = PROGRESS_COMMAND_VACUUM;
  else if (pg_strcasecmp(cmd, "ANALYZE") == 0)
    cmdtype = PROGRESS_COMMAND_ANALYZE;
  else if (pg_strcasecmp(cmd, "CLUSTER") == 0)
    cmdtype = PROGRESS_COMMAND_CLUSTER;
  else if (pg_strcasecmp(cmd, "CREATE INDEX") == 0)
    cmdtype = PROGRESS_COMMAND_CREATE_INDEX;
  else if (pg_strcasecmp(cmd, "BASEBACKUP") == 0)
    cmdtype = PROGRESS_COMMAND_BASEBACKUP;
  else if (pg_strcasecmp(cmd, "COPY") == 0)
    cmdtype = PROGRESS_COMMAND_COPY;
  else
    ereport(ERROR,
        (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
         errmsg("invalid command name: \"%s\"", cmd)));

  InitMaterializedSRF(fcinfo, 0);

  /* 1-based index */
  for (curr_backend = 1; curr_backend <= num_backends; curr_backend++)
  {
    LocalPgBackendStatus *local_beentry;
    PgBackendStatus *beentry;
    Datum   values[PG_STAT_GET_PROGRESS_COLS] = {0};
    bool    nulls[PG_STAT_GET_PROGRESS_COLS] = {0};
    int     i;

    local_beentry = pgstat_get_local_beentry_by_index(curr_backend);
    beentry = &local_beentry->backendStatus;

    /*
     * Report values for only those backends which are running the given
     * command.
     */
    if (beentry->st_progress_command != cmdtype)
      continue;

    /* Value available to all callers */
    values[0] = Int32GetDatum(beentry->st_procpid);
    values[1] = ObjectIdGetDatum(beentry->st_databaseid);

    /* show rest of the values including relid only to role members */
    if (HAS_PGSTAT_PERMISSIONS(beentry->st_userid))
    {
      values[2] = ObjectIdGetDatum(beentry->st_progress_command_target);
      for (i = 0; i < PGSTAT_NUM_PROGRESS_PARAM; i++)
        values[i + 3] = Int64GetDatum(beentry->st_progress_param[i]);
    }
    else
    {
      nulls[2] = true;
      for (i = 0; i < PGSTAT_NUM_PROGRESS_PARAM; i++)
        nulls[i + 3] = true;
    }

    tuplestore_putvalues(rsinfo->setResult, rsinfo->setDesc, values, nulls);
  }

  return (Datum) 0;
}

/*
 * Returns activity of PG backends.
 */
Datum
pg_stat_get_activity(PG_FUNCTION_ARGS)
{
#define PG_STAT_GET_ACTIVITY_COLS 31
  int     num_backends = pgstat_fetch_stat_numbackends();
  int     curr_backend;
  int     pid = PG_ARGISNULL(0) ? -1 : PG_GETARG_INT32(0);
  ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;

  InitMaterializedSRF(fcinfo, 0);

  /* 1-based index */
  for (curr_backend = 1; curr_backend <= num_backends; curr_backend++)
  {
    /* for each row */
    Datum   values[PG_STAT_GET_ACTIVITY_COLS] = {0};
    bool    nulls[PG_STAT_GET_ACTIVITY_COLS] = {0};
    LocalPgBackendStatus *local_beentry;
    PgBackendStatus *beentry;
    PGPROC     *proc;
    const char *wait_event_type = NULL;
    const char *wait_event = NULL;

    /* Get the next one in the list */
    local_beentry = pgstat_get_local_beentry_by_index(curr_backend);
    beentry = &local_beentry->backendStatus;

    /* If looking for specific PID, ignore all the others */
    if (pid != -1 && beentry->st_procpid != pid)
      continue;

    /* Values available to all callers */
    if (beentry->st_databaseid != InvalidOid)
      values[0] = ObjectIdGetDatum(beentry->st_databaseid);
    else
      nulls[0] = true;

    values[1] = Int32GetDatum(beentry->st_procpid);

    if (beentry->st_userid != InvalidOid)
      values[2] = ObjectIdGetDatum(beentry->st_userid);
    else
      nulls[2] = true;

    if (beentry->st_appname)
      values[3] = CStringGetTextDatum(beentry->st_appname);
    else
      nulls[3] = true;

    if (TransactionIdIsValid(local_beentry->backend_xid))
      values[15] = TransactionIdGetDatum(local_beentry->backend_xid);
    else
      nulls[15] = true;

    if (TransactionIdIsValid(local_beentry->backend_xmin))
      values[16] = TransactionIdGetDatum(local_beentry->backend_xmin);
    else
      nulls[16] = true;

    /* Values only available to role member or pg_read_all_stats */
    if (HAS_PGSTAT_PERMISSIONS(beentry->st_userid))
    {
      char     *clipped_activity;

      switch (beentry->st_state)
      {
        case STATE_STARTING:
          values[4] = CStringGetTextDatum("starting");
          break;
        case STATE_IDLE:
          values[4] = CStringGetTextDatum("idle");
          break;
        case STATE_RUNNING:
          values[4] = CStringGetTextDatum("active");
          break;
        case STATE_IDLEINTRANSACTION:
          values[4] = CStringGetTextDatum("idle in transaction");
          break;
        case STATE_FASTPATH:
          values[4] = CStringGetTextDatum("fastpath function call");
          break;
        case STATE_IDLEINTRANSACTION_ABORTED:
          values[4] = CStringGetTextDatum("idle in transaction (aborted)");
          break;
        case STATE_DISABLED:
          values[4] = CStringGetTextDatum("disabled");
          break;
        case STATE_UNDEFINED:
          nulls[4] = true;
          break;
      }

      clipped_activity = pgstat_clip_activity(beentry->st_activity_raw);
      values[5] = CStringGetTextDatum(clipped_activity);
      pfree(clipped_activity);

      /* leader_pid */
      nulls[29] = true;

      proc = BackendPidGetProc(beentry->st_procpid);

      if (proc == NULL && (beentry->st_backendType != B_BACKEND))
      {
        /*
         * For an auxiliary process, retrieve process info from
         * AuxiliaryProcs stored in shared-memory.
         */
        proc = AuxiliaryPidGetProc(beentry->st_procpid);
      }

      /*
       * If a PGPROC entry was retrieved, display wait events and lock
       * group leader or apply leader information if any.  To avoid
       * extra overhead, no extra lock is being held, so there is no
       * guarantee of consistency across multiple rows.
       */
      if (proc != NULL)
      {
        uint32    raw_wait_event;
        PGPROC     *leader;

        raw_wait_event = UINT32_ACCESS_ONCE(proc->wait_event_info);
        wait_event_type = pgstat_get_wait_event_type(raw_wait_event);
        wait_event = pgstat_get_wait_event(raw_wait_event);

        leader = proc->lockGroupLeader;

        /*
         * Show the leader only for active parallel workers.  This
         * leaves the field as NULL for the leader of a parallel group
         * or the leader of parallel apply workers.
         */
        if (leader && leader->pid != beentry->st_procpid)
        {
          values[29] = Int32GetDatum(leader->pid);
          nulls[29] = false;
        }
        else if (beentry->st_backendType == B_BG_WORKER)
        {
          int     leader_pid = GetLeaderApplyWorkerPid(beentry->st_procpid);

          if (leader_pid != InvalidPid)
          {
            values[29] = Int32GetDatum(leader_pid);
            nulls[29] = false;
          }
        }
      }

      if (wait_event_type)
        values[6] = CStringGetTextDatum(wait_event_type);
      else
        nulls[6] = true;

      if (wait_event)
        values[7] = CStringGetTextDatum(wait_event);
      else
        nulls[7] = true;

      /*
       * Don't expose transaction time for walsenders; it confuses
       * monitoring, particularly because we don't keep the time up-to-
       * date.
       */
      if (beentry->st_xact_start_timestamp != 0 &&
        beentry->st_backendType != B_WAL_SENDER)
        values[8] = TimestampTzGetDatum(beentry->st_xact_start_timestamp);
      else
        nulls[8] = true;

      if (beentry->st_activity_start_timestamp != 0)
        values[9] = TimestampTzGetDatum(beentry->st_activity_start_timestamp);
      else
        nulls[9] = true;

      if (beentry->st_proc_start_timestamp != 0)
        values[10] = TimestampTzGetDatum(beentry->st_proc_start_timestamp);
      else
        nulls[10] = true;

      if (beentry->st_state_start_timestamp != 0)
        values[11] = TimestampTzGetDatum(beentry->st_state_start_timestamp);
      else
        nulls[11] = true;

      /* A zeroed client addr means we don't know */
      if (pg_memory_is_all_zeros(&beentry->st_clientaddr,
                     sizeof(beentry->st_clientaddr)))
      {
        nulls[12] = true;
        nulls[13] = true;
        nulls[14] = true;
      }
      else
      {
        if (beentry->st_clientaddr.addr.ss_family == AF_INET ||
          beentry->st_clientaddr.addr.ss_family == AF_INET6)
        {
          char    remote_host[NI_MAXHOST];
          char    remote_port[NI_MAXSERV];
          int     ret;

          remote_host[0] = '\0';
          remote_port[0] = '\0';
          ret = pg_getnameinfo_all(&beentry->st_clientaddr.addr,
                       beentry->st_clientaddr.salen,
                       remote_host, sizeof(remote_host),
                       remote_port, sizeof(remote_port),
                       NI_NUMERICHOST | NI_NUMERICSERV);
          if (ret == 0)
          {
            clean_ipv6_addr(beentry->st_clientaddr.addr.ss_family, remote_host);
            values[12] = DirectFunctionCall1(inet_in,
                             CStringGetDatum(remote_host));
            if (beentry->st_clienthostname &&
              beentry->st_clienthostname[0])
              values[13] = CStringGetTextDatum(beentry->st_clienthostname);
            else
              nulls[13] = true;
            values[14] = Int32GetDatum(atoi(remote_port));
          }
          else
          {
            nulls[12] = true;
            nulls[13] = true;
            nulls[14] = true;
          }
        }
        else if (beentry->st_clientaddr.addr.ss_family == AF_UNIX)
        {
          /*
           * Unix sockets always reports NULL for host and -1 for
           * port, so it's possible to tell the difference to
           * connections we have no permissions to view, or with
           * errors.
           */
          nulls[12] = true;
          nulls[13] = true;
          values[14] = Int32GetDatum(-1);
        }
        else
        {
          /* Unknown address type, should never happen */
          nulls[12] = true;
          nulls[13] = true;
          nulls[14] = true;
        }
      }
      /* Add backend type */
      if (beentry->st_backendType == B_BG_WORKER)
      {
        const char *bgw_type;

        bgw_type = GetBackgroundWorkerTypeByPid(beentry->st_procpid);
        if (bgw_type)
          values[17] = CStringGetTextDatum(bgw_type);
        else
          nulls[17] = true;
      }
      else
        values[17] =
          CStringGetTextDatum(GetBackendTypeDesc(beentry->st_backendType));

      /* SSL information */
      if (beentry->st_ssl)
      {
        values[18] = BoolGetDatum(true);  /* ssl */
        values[19] = CStringGetTextDatum(beentry->st_sslstatus->ssl_version);
        values[20] = CStringGetTextDatum(beentry->st_sslstatus->ssl_cipher);
        values[21] = Int32GetDatum(beentry->st_sslstatus->ssl_bits);

        if (beentry->st_sslstatus->ssl_client_dn[0])
          values[22] = CStringGetTextDatum(beentry->st_sslstatus->ssl_client_dn);
        else
          nulls[22] = true;

        if (beentry->st_sslstatus->ssl_client_serial[0])
          values[23] = DirectFunctionCall3(numeric_in,
                           CStringGetDatum(beentry->st_sslstatus->ssl_client_serial),
                           ObjectIdGetDatum(InvalidOid),
                           Int32GetDatum(-1));
        else
          nulls[23] = true;

        if (beentry->st_sslstatus->ssl_issuer_dn[0])
          values[24] = CStringGetTextDatum(beentry->st_sslstatus->ssl_issuer_dn);
        else
          nulls[24] = true;
      }
      else
      {
        values[18] = BoolGetDatum(false); /* ssl */
        nulls[19] = nulls[20] = nulls[21] = nulls[22] = nulls[23] = nulls[24] = true;
      }

      /* GSSAPI information */
      if (beentry->st_gss)
      {
        values[25] = BoolGetDatum(beentry->st_gssstatus->gss_auth); /* gss_auth */
        values[26] = CStringGetTextDatum(beentry->st_gssstatus->gss_princ);
        values[27] = BoolGetDatum(beentry->st_gssstatus->gss_enc);  /* GSS Encryption in use */
        values[28] = BoolGetDatum(beentry->st_gssstatus->gss_delegation); /* GSS credentials
                                           * delegated */
      }
      else
      {
        values[25] = BoolGetDatum(false); /* gss_auth */
        nulls[26] = true; /* No GSS principal */
        values[27] = BoolGetDatum(false); /* GSS Encryption not in
                           * use */
        values[28] = BoolGetDatum(false); /* GSS credentials not
                           * delegated */
      }
      if (beentry->st_query_id == INT64CONST(0))
        nulls[30] = true;
      else
        values[30] = Int64GetDatum(beentry->st_query_id);
    }
    else
    {
      /* No permissions to view data about this session */
      values[5] = CStringGetTextDatum("<insufficient privilege>");
      nulls[4] = true;
      nulls[6] = true;
      nulls[7] = true;
      nulls[8] = true;
      nulls[9] = true;
      nulls[10] = true;
      nulls[11] = true;
      nulls[12] = true;
      nulls[13] = true;
      nulls[14] = true;
      nulls[17] = true;
      nulls[18] = true;
      nulls[19] = true;
      nulls[20] = true;
      nulls[21] = true;
      nulls[22] = true;
      nulls[23] = true;
      nulls[24] = true;
      nulls[25] = true;
      nulls[26] = true;
      nulls[27] = true;
      nulls[28] = true;
      nulls[29] = true;
      nulls[30] = true;
    }

    tuplestore_putvalues(rsinfo->setResult, rsinfo->setDesc, values, nulls);

    /* If only a single backend was requested, and we found it, break. */
    if (pid != -1)
      break;
  }

  return (Datum) 0;
}


Datum
pg_backend_pid(PG_FUNCTION_ARGS)
{
  PG_RETURN_INT32(MyProcPid);
}


Datum
pg_stat_get_backend_pid(PG_FUNCTION_ARGS)
{
  int32   procNumber = PG_GETARG_INT32(0);
  PgBackendStatus *beentry;

  if ((beentry = pgstat_get_beentry_by_proc_number(procNumber)) == NULL)
    PG_RETURN_NULL();

  PG_RETURN_INT32(beentry->st_procpid);
}


Datum
pg_stat_get_backend_dbid(PG_FUNCTION_ARGS)
{
  int32   procNumber = PG_GETARG_INT32(0);
  PgBackendStatus *beentry;

  if ((beentry = pgstat_get_beentry_by_proc_number(procNumber)) == NULL)
    PG_RETURN_NULL();

  PG_RETURN_OID(beentry->st_databaseid);
}


Datum
pg_stat_get_backend_userid(PG_FUNCTION_ARGS)
{
  int32   procNumber = PG_GETARG_INT32(0);
  PgBackendStatus *beentry;

  if ((beentry = pgstat_get_beentry_by_proc_number(procNumber)) == NULL)
    PG_RETURN_NULL();

  PG_RETURN_OID(beentry->st_userid);
}

Datum
pg_stat_get_backend_subxact(PG_FUNCTION_ARGS)
{
#define PG_STAT_GET_SUBXACT_COLS  2
  TupleDesc tupdesc;
  Datum   values[PG_STAT_GET_SUBXACT_COLS] = {0};
  bool    nulls[PG_STAT_GET_SUBXACT_COLS] = {0};
  int32   procNumber = PG_GETARG_INT32(0);
  LocalPgBackendStatus *local_beentry;

  /* Initialise attributes information in the tuple descriptor */
  tupdesc = CreateTemplateTupleDesc(PG_STAT_GET_SUBXACT_COLS);
  TupleDescInitEntry(tupdesc, (AttrNumber) 1, "subxact_count",
             INT4OID, -1, 0);
  TupleDescInitEntry(tupdesc, (AttrNumber) 2, "subxact_overflow",
             BOOLOID, -1, 0);

  BlessTupleDesc(tupdesc);

  if ((local_beentry = pgstat_get_local_beentry_by_proc_number(procNumber)) != NULL)
  {
    /* Fill values and NULLs */
    values[0] = Int32GetDatum(local_beentry->backend_subxact_count);
    values[1] = BoolGetDatum(local_beentry->backend_subxact_overflowed);
  }
  else
  {
    nulls[0] = true;
    nulls[1] = true;
  }

  /* Returns the record as Datum */
  PG_RETURN_DATUM(HeapTupleGetDatum(heap_form_tuple(tupdesc, values, nulls)));
}

Datum
pg_stat_get_backend_activity(PG_FUNCTION_ARGS)
{
  int32   procNumber = PG_GETARG_INT32(0);
  PgBackendStatus *beentry;
  const char *activity;
  char     *clipped_activity;
  text     *ret;

  if ((beentry = pgstat_get_beentry_by_proc_number(procNumber)) == NULL)
    activity = "<backend information not available>";
  else if (!HAS_PGSTAT_PERMISSIONS(beentry->st_userid))
    activity = "<insufficient privilege>";
  else if (*(beentry->st_activity_raw) == '\0')
    activity = "<command string not enabled>";
  else
    activity = beentry->st_activity_raw;

  clipped_activity = pgstat_clip_activity(activity);
  ret = cstring_to_text(activity);
  pfree(clipped_activity);

  PG_RETURN_TEXT_P(ret);
}

Datum
pg_stat_get_backend_wait_event_type(PG_FUNCTION_ARGS)
{
  int32   procNumber = PG_GETARG_INT32(0);
  PgBackendStatus *beentry;
  PGPROC     *proc;
  const char *wait_event_type = NULL;

  if ((beentry = pgstat_get_beentry_by_proc_number(procNumber)) == NULL)
    wait_event_type = "<backend information not available>";
  else if (!HAS_PGSTAT_PERMISSIONS(beentry->st_userid))
    wait_event_type = "<insufficient privilege>";
  else if ((proc = BackendPidGetProc(beentry->st_procpid)) != NULL)
    wait_event_type = pgstat_get_wait_event_type(proc->wait_event_info);

  if (!wait_event_type)
    PG_RETURN_NULL();

  PG_RETURN_TEXT_P(cstring_to_text(wait_event_type));
}

Datum
pg_stat_get_backend_wait_event(PG_FUNCTION_ARGS)
{
  int32   procNumber = PG_GETARG_INT32(0);
  PgBackendStatus *beentry;
  PGPROC     *proc;
  const char *wait_event = NULL;

  if ((beentry = pgstat_get_beentry_by_proc_number(procNumber)) == NULL)
    wait_event = "<backend information not available>";
  else if (!HAS_PGSTAT_PERMISSIONS(beentry->st_userid))
    wait_event = "<insufficient privilege>";
  else if ((proc = BackendPidGetProc(beentry->st_procpid)) != NULL)
    wait_event = pgstat_get_wait_event(proc->wait_event_info);

  if (!wait_event)
    PG_RETURN_NULL();

  PG_RETURN_TEXT_P(cstring_to_text(wait_event));
}


Datum
pg_stat_get_backend_activity_start(PG_FUNCTION_ARGS)
{
  int32   procNumber = PG_GETARG_INT32(0);
  TimestampTz result;
  PgBackendStatus *beentry;

  if ((beentry = pgstat_get_beentry_by_proc_number(procNumber)) == NULL)
    PG_RETURN_NULL();

  else if (!HAS_PGSTAT_PERMISSIONS(beentry->st_userid))
    PG_RETURN_NULL();

  result = beentry->st_activity_start_timestamp;

  /*
   * No time recorded for start of current query -- this is the case if the
   * user hasn't enabled query-level stats collection.
   */
  if (result == 0)
    PG_RETURN_NULL();

  PG_RETURN_TIMESTAMPTZ(result);
}


Datum
pg_stat_get_backend_xact_start(PG_FUNCTION_ARGS)
{
  int32   procNumber = PG_GETARG_INT32(0);
  TimestampTz result;
  PgBackendStatus *beentry;

  if ((beentry = pgstat_get_beentry_by_proc_number(procNumber)) == NULL)
    PG_RETURN_NULL();

  else if (!HAS_PGSTAT_PERMISSIONS(beentry->st_userid))
    PG_RETURN_NULL();

  result = beentry->st_xact_start_timestamp;

  if (result == 0)     /* not in a transaction */
    PG_RETURN_NULL();

  PG_RETURN_TIMESTAMPTZ(result);
}


Datum
pg_stat_get_backend_start(PG_FUNCTION_ARGS)
{
  int32   procNumber = PG_GETARG_INT32(0);
  TimestampTz result;
  PgBackendStatus *beentry;

  if ((beentry = pgstat_get_beentry_by_proc_number(procNumber)) == NULL)
    PG_RETURN_NULL();

  else if (!HAS_PGSTAT_PERMISSIONS(beentry->st_userid))
    PG_RETURN_NULL();

  result = beentry->st_proc_start_timestamp;

  if (result == 0)     /* probably can't happen? */
    PG_RETURN_NULL();

  PG_RETURN_TIMESTAMPTZ(result);
}


Datum
pg_stat_get_backend_client_addr(PG_FUNCTION_ARGS)
{
  int32   procNumber = PG_GETARG_INT32(0);
  PgBackendStatus *beentry;
  char    remote_host[NI_MAXHOST];
  int     ret;

  if ((beentry = pgstat_get_beentry_by_proc_number(procNumber)) == NULL)
    PG_RETURN_NULL();

  else if (!HAS_PGSTAT_PERMISSIONS(beentry->st_userid))
    PG_RETURN_NULL();

  /* A zeroed client addr means we don't know */
  if (pg_memory_is_all_zeros(&beentry->st_clientaddr,
                 sizeof(beentry->st_clientaddr)))
    PG_RETURN_NULL();

  switch (beentry->st_clientaddr.addr.ss_family)
  {
    case AF_INET:
    case AF_INET6:
      break;
    default:
      PG_RETURN_NULL();
  }

  remote_host[0] = '\0';
  ret = pg_getnameinfo_all(&beentry->st_clientaddr.addr,
               beentry->st_clientaddr.salen,
               remote_host, sizeof(remote_host),
               NULL, 0,
               NI_NUMERICHOST | NI_NUMERICSERV);
  if (ret != 0)
    PG_RETURN_NULL();

  clean_ipv6_addr(beentry->st_clientaddr.addr.ss_family, remote_host);

  PG_RETURN_DATUM(DirectFunctionCall1(inet_in,
                    CStringGetDatum(remote_host)));
}

Datum
pg_stat_get_backend_client_port(PG_FUNCTION_ARGS)
{
  int32   procNumber = PG_GETARG_INT32(0);
  PgBackendStatus *beentry;
  char    remote_port[NI_MAXSERV];
  int     ret;

  if ((beentry = pgstat_get_beentry_by_proc_number(procNumber)) == NULL)
    PG_RETURN_NULL();

  else if (!HAS_PGSTAT_PERMISSIONS(beentry->st_userid))
    PG_RETURN_NULL();

  /* A zeroed client addr means we don't know */
  if (pg_memory_is_all_zeros(&beentry->st_clientaddr,
                 sizeof(beentry->st_clientaddr)))
    PG_RETURN_NULL();

  switch (beentry->st_clientaddr.addr.ss_family)
  {
    case AF_INET:
    case AF_INET6:
      break;
    case AF_UNIX:
      PG_RETURN_INT32(-1);
    default:
      PG_RETURN_NULL();
  }

  remote_port[0] = '\0';
  ret = pg_getnameinfo_all(&beentry->st_clientaddr.addr,
               beentry->st_clientaddr.salen,
               NULL, 0,
               remote_port, sizeof(remote_port),
               NI_NUMERICHOST | NI_NUMERICSERV);
  if (ret != 0)
    PG_RETURN_NULL();

  PG_RETURN_DATUM(DirectFunctionCall1(int4in,
                    CStringGetDatum(remote_port)));
}


Datum
pg_stat_get_db_numbackends(PG_FUNCTION_ARGS)
{
  Oid     dbid = PG_GETARG_OID(0);
  int32   result;
  int     tot_backends = pgstat_fetch_stat_numbackends();
  int     idx;

  result = 0;
  for (idx = 1; idx <= tot_backends; idx++)
  {
    LocalPgBackendStatus *local_beentry = pgstat_get_local_beentry_by_index(idx);

    if (local_beentry->backendStatus.st_databaseid == dbid)
      result++;
  }

  PG_RETURN_INT32(result);
}


#define PG_STAT_GET_DBENTRY_INT64(stat)             \
Datum                             \
CppConcat(pg_stat_get_db_,stat)(PG_FUNCTION_ARGS)       \
{                               \
  Oid     dbid = PG_GETARG_OID(0);            \
  int64   result;                     \
  PgStat_StatDBEntry *dbentry;                \
                                \
  if ((dbentry = pgstat_fetch_stat_dbentry(dbid)) == NULL) \
    result = 0;                       \
  else                            \
    result = (int64) (dbentry->stat);           \
                                \
  PG_RETURN_INT64(result);                 \
}

/* pg_stat_get_db_blocks_fetched */
PG_STAT_GET_DBENTRY_INT64(blocks_fetched)

/* pg_stat_get_db_blocks_hit */
PG_STAT_GET_DBENTRY_INT64(blocks_hit)

/* pg_stat_get_db_conflict_bufferpin */
PG_STAT_GET_DBENTRY_INT64(conflict_bufferpin)

/* pg_stat_get_db_conflict_lock */
PG_STAT_GET_DBENTRY_INT64(conflict_lock)

/* pg_stat_get_db_conflict_snapshot */
PG_STAT_GET_DBENTRY_INT64(conflict_snapshot)

/* pg_stat_get_db_conflict_startup_deadlock */
PG_STAT_GET_DBENTRY_INT64(conflict_startup_deadlock)

/* pg_stat_get_db_conflict_tablespace */
PG_STAT_GET_DBENTRY_INT64(conflict_tablespace)

/* pg_stat_get_db_deadlocks */
PG_STAT_GET_DBENTRY_INT64(deadlocks)

/* pg_stat_get_db_sessions */
PG_STAT_GET_DBENTRY_INT64(sessions)

/* pg_stat_get_db_sessions_abandoned */
PG_STAT_GET_DBENTRY_INT64(sessions_abandoned)

/* pg_stat_get_db_sessions_fatal */
PG_STAT_GET_DBENTRY_INT64(sessions_fatal)

/* pg_stat_get_db_sessions_killed */
PG_STAT_GET_DBENTRY_INT64(sessions_killed)

/* pg_stat_get_db_parallel_workers_to_launch */
PG_STAT_GET_DBENTRY_INT64(parallel_workers_to_launch)

/* pg_stat_get_db_parallel_workers_launched */
PG_STAT_GET_DBENTRY_INT64(parallel_workers_launched)

/* pg_stat_get_db_temp_bytes */
PG_STAT_GET_DBENTRY_INT64(temp_bytes)

/* pg_stat_get_db_temp_files */
PG_STAT_GET_DBENTRY_INT64(temp_files)

/* pg_stat_get_db_tuples_deleted */
PG_STAT_GET_DBENTRY_INT64(tuples_deleted)

/* pg_stat_get_db_tuples_fetched */
PG_STAT_GET_DBENTRY_INT64(tuples_fetched)

/* pg_stat_get_db_tuples_inserted */
PG_STAT_GET_DBENTRY_INT64(tuples_inserted)

/* pg_stat_get_db_tuples_returned */
PG_STAT_GET_DBENTRY_INT64(tuples_returned)

/* pg_stat_get_db_tuples_updated */
PG_STAT_GET_DBENTRY_INT64(tuples_updated)

/* pg_stat_get_db_xact_commit */
PG_STAT_GET_DBENTRY_INT64(xact_commit)

/* pg_stat_get_db_xact_rollback */
PG_STAT_GET_DBENTRY_INT64(xact_rollback)

/* pg_stat_get_db_conflict_logicalslot */
PG_STAT_GET_DBENTRY_INT64(conflict_logicalslot)

Datum
pg_stat_get_db_stat_reset_time(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->stat_reset_timestamp;

  if (result == 0)
    PG_RETURN_NULL();
  else
    PG_RETURN_TIMESTAMPTZ(result);
}


Datum
pg_stat_get_db_conflict_all(PG_FUNCTION_ARGS)
{
  Oid     dbid = PG_GETARG_OID(0);
  int64   result;
  PgStat_StatDBEntry *dbentry;

  if ((dbentry = pgstat_fetch_stat_dbentry(dbid)) == NULL)
    result = 0;
  else
    result = (int64) (dbentry->conflict_tablespace +
              dbentry->conflict_lock +
              dbentry->conflict_snapshot +
              dbentry->conflict_logicalslot +
              dbentry->conflict_bufferpin +
              dbentry->conflict_startup_deadlock);

  PG_RETURN_INT64(result);
}

Datum
pg_stat_get_db_checksum_failures(PG_FUNCTION_ARGS)
{
  Oid     dbid = PG_GETARG_OID(0);
  int64   result;
  PgStat_StatDBEntry *dbentry;

  if (!DataChecksumsEnabled())
    PG_RETURN_NULL();

  if ((dbentry = pgstat_fetch_stat_dbentry(dbid)) == NULL)
    result = 0;
  else
    result = (int64) (dbentry->checksum_failures);

  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 (!DataChecksumsEnabled())
    PG_RETURN_NULL();

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

/* convert counter from microsec to millisec for display */
#define PG_STAT_GET_DBENTRY_FLOAT8_MS(stat)           \
Datum                             \
CppConcat(pg_stat_get_db_,stat)(PG_FUNCTION_ARGS)       \
{                               \
  Oid     dbid = PG_GETARG_OID(0);            \
  double    result;                     \
  PgStat_StatDBEntry *dbentry;                \
                                \
  if ((dbentry = pgstat_fetch_stat_dbentry(dbid)) == NULL) \
    result = 0;                       \
  else                            \
    result = ((double) dbentry->stat) / 1000.0;       \
                                \
  PG_RETURN_FLOAT8(result);                 \
}

/* pg_stat_get_db_active_time */
PG_STAT_GET_DBENTRY_FLOAT8_MS(active_time)

/* pg_stat_get_db_blk_read_time */
PG_STAT_GET_DBENTRY_FLOAT8_MS(blk_read_time)

/* pg_stat_get_db_blk_write_time */
PG_STAT_GET_DBENTRY_FLOAT8_MS(blk_write_time)

/* pg_stat_get_db_idle_in_transaction_time */
PG_STAT_GET_DBENTRY_FLOAT8_MS(idle_in_transaction_time)

/* pg_stat_get_db_session_time */
PG_STAT_GET_DBENTRY_FLOAT8_MS(session_time)

Datum
pg_stat_get_checkpointer_num_timed(PG_FUNCTION_ARGS)
{
  PG_RETURN_INT64(pgstat_fetch_stat_checkpointer()->num_timed);
}

Datum
pg_stat_get_checkpointer_num_requested(PG_FUNCTION_ARGS)
{
  PG_RETURN_INT64(pgstat_fetch_stat_checkpointer()->num_requested);
}

Datum
pg_stat_get_checkpointer_num_performed(PG_FUNCTION_ARGS)
{
  PG_RETURN_INT64(pgstat_fetch_stat_checkpointer()->num_performed);
}

Datum
pg_stat_get_checkpointer_restartpoints_timed(PG_FUNCTION_ARGS)
{
  PG_RETURN_INT64(pgstat_fetch_stat_checkpointer()->restartpoints_timed);
}

Datum
pg_stat_get_checkpointer_restartpoints_requested(PG_FUNCTION_ARGS)
{
  PG_RETURN_INT64(pgstat_fetch_stat_checkpointer()->restartpoints_requested);
}

Datum
pg_stat_get_checkpointer_restartpoints_performed(PG_FUNCTION_ARGS)
{
  PG_RETURN_INT64(pgstat_fetch_stat_checkpointer()->restartpoints_performed);
}

Datum
pg_stat_get_checkpointer_buffers_written(PG_FUNCTION_ARGS)
{
  PG_RETURN_INT64(pgstat_fetch_stat_checkpointer()->buffers_written);
}

Datum
pg_stat_get_checkpointer_slru_written(PG_FUNCTION_ARGS)
{
  PG_RETURN_INT64(pgstat_fetch_stat_checkpointer()->slru_written);
}

Datum
pg_stat_get_bgwriter_buf_written_clean(PG_FUNCTION_ARGS)
{
  PG_RETURN_INT64(pgstat_fetch_stat_bgwriter()->buf_written_clean);
}

Datum
pg_stat_get_bgwriter_maxwritten_clean(PG_FUNCTION_ARGS)
{
  PG_RETURN_INT64(pgstat_fetch_stat_bgwriter()->maxwritten_clean);
}

Datum
pg_stat_get_checkpointer_write_time(PG_FUNCTION_ARGS)
{
  /* time is already in msec, just convert to double for presentation */
  PG_RETURN_FLOAT8((double)
           pgstat_fetch_stat_checkpointer()->write_time);
}

Datum
pg_stat_get_checkpointer_sync_time(PG_FUNCTION_ARGS)
{
  /* time is already in msec, just convert to double for presentation */
  PG_RETURN_FLOAT8((double)
           pgstat_fetch_stat_checkpointer()->sync_time);
}

Datum
pg_stat_get_checkpointer_stat_reset_time(PG_FUNCTION_ARGS)
{
  PG_RETURN_TIMESTAMPTZ(pgstat_fetch_stat_checkpointer()->stat_reset_timestamp);
}

Datum
pg_stat_get_bgwriter_stat_reset_time(PG_FUNCTION_ARGS)
{
  PG_RETURN_TIMESTAMPTZ(pgstat_fetch_stat_bgwriter()->stat_reset_timestamp);
}

Datum
pg_stat_get_buf_alloc(PG_FUNCTION_ARGS)
{
  PG_RETURN_INT64(pgstat_fetch_stat_bgwriter()->buf_alloc);
}

/*
* When adding a new column to the pg_stat_io view and the
* pg_stat_get_backend_io() function, add a new enum value here above
* IO_NUM_COLUMNS.
*/
typedef enum io_stat_col
{
  IO_COL_INVALID = -1,
  IO_COL_BACKEND_TYPE,
  IO_COL_OBJECT,
  IO_COL_CONTEXT,
  IO_COL_READS,
  IO_COL_READ_BYTES,
  IO_COL_READ_TIME,
  IO_COL_WRITES,
  IO_COL_WRITE_BYTES,
  IO_COL_WRITE_TIME,
  IO_COL_WRITEBACKS,
  IO_COL_WRITEBACK_TIME,
  IO_COL_EXTENDS,
  IO_COL_EXTEND_BYTES,
  IO_COL_EXTEND_TIME,
  IO_COL_HITS,
  IO_COL_EVICTIONS,
  IO_COL_REUSES,
  IO_COL_FSYNCS,
  IO_COL_FSYNC_TIME,
  IO_COL_RESET_TIME,
  IO_NUM_COLUMNS,
} io_stat_col;

/*
 * When adding a new IOOp, add a new io_stat_col and add a case to this
 * function returning the corresponding io_stat_col.
 */
static io_stat_col
pgstat_get_io_op_index(IOOp io_op)
{
  switch (io_op)
  {
    case IOOP_EVICT:
      return IO_COL_EVICTIONS;
    case IOOP_EXTEND:
      return IO_COL_EXTENDS;
    case IOOP_FSYNC:
      return IO_COL_FSYNCS;
    case IOOP_HIT:
      return IO_COL_HITS;
    case IOOP_READ:
      return IO_COL_READS;
    case IOOP_REUSE:
      return IO_COL_REUSES;
    case IOOP_WRITE:
      return IO_COL_WRITES;
    case IOOP_WRITEBACK:
      return IO_COL_WRITEBACKS;
  }

  elog(ERROR, "unrecognized IOOp value: %d", io_op);
  pg_unreachable();
}

/*
 * Get the number of the column containing IO bytes for the specified IOOp.
 * If an IOOp is not tracked in bytes, IO_COL_INVALID is returned.
 */
static io_stat_col
pgstat_get_io_byte_index(IOOp io_op)
{
  switch (io_op)
  {
    case IOOP_EXTEND:
      return IO_COL_EXTEND_BYTES;
    case IOOP_READ:
      return IO_COL_READ_BYTES;
    case IOOP_WRITE:
      return IO_COL_WRITE_BYTES;
    case IOOP_EVICT:
    case IOOP_FSYNC:
    case IOOP_HIT:
    case IOOP_REUSE:
    case IOOP_WRITEBACK:
      return IO_COL_INVALID;
  }

  elog(ERROR, "unrecognized IOOp value: %d", io_op);
  pg_unreachable();
}

/*
 * Get the number of the column containing IO times for the specified IOOp.
 * If an op has no associated time, IO_COL_INVALID is returned.
 */
static io_stat_col
pgstat_get_io_time_index(IOOp io_op)
{
  switch (io_op)
  {
    case IOOP_READ:
      return IO_COL_READ_TIME;
    case IOOP_WRITE:
      return IO_COL_WRITE_TIME;
    case IOOP_WRITEBACK:
      return IO_COL_WRITEBACK_TIME;
    case IOOP_EXTEND:
      return IO_COL_EXTEND_TIME;
    case IOOP_FSYNC:
      return IO_COL_FSYNC_TIME;
    case IOOP_EVICT:
    case IOOP_HIT:
    case IOOP_REUSE:
      return IO_COL_INVALID;
  }

  elog(ERROR, "unrecognized IOOp value: %d", io_op);
  pg_unreachable();
}

static inline double
pg_stat_us_to_ms(PgStat_Counter val_ms)
{
  return val_ms * (double) 0.001;
}

/*
 * pg_stat_io_build_tuples
 *
 * Helper routine for pg_stat_get_io() and pg_stat_get_backend_io()
 * filling a result tuplestore with one tuple for each object and each
 * context supported by the caller, based on the contents of bktype_stats.
 */
static void
pg_stat_io_build_tuples(ReturnSetInfo *rsinfo,
            PgStat_BktypeIO *bktype_stats,
            BackendType bktype,
            TimestampTz stat_reset_timestamp)
{
  Datum   bktype_desc = CStringGetTextDatum(GetBackendTypeDesc(bktype));

  for (int io_obj = 0; io_obj < IOOBJECT_NUM_TYPES; io_obj++)
  {
    const char *obj_name = pgstat_get_io_object_name(io_obj);

    for (int io_context = 0; io_context < IOCONTEXT_NUM_TYPES; io_context++)
    {
      const char *context_name = pgstat_get_io_context_name(io_context);

      Datum   values[IO_NUM_COLUMNS] = {0};
      bool    nulls[IO_NUM_COLUMNS] = {0};

      /*
       * Some combinations of BackendType, IOObject, and IOContext are
       * not valid for any type of IOOp. In such cases, omit the entire
       * row from the view.
       */
      if (!pgstat_tracks_io_object(bktype, io_obj, io_context))
        continue;

      values[IO_COL_BACKEND_TYPE] = bktype_desc;
      values[IO_COL_CONTEXT] = CStringGetTextDatum(context_name);
      values[IO_COL_OBJECT] = CStringGetTextDatum(obj_name);
      if (stat_reset_timestamp != 0)
        values[IO_COL_RESET_TIME] = TimestampTzGetDatum(stat_reset_timestamp);
      else
        nulls[IO_COL_RESET_TIME] = true;

      for (int io_op = 0; io_op < IOOP_NUM_TYPES; io_op++)
      {
        int     op_idx = pgstat_get_io_op_index(io_op);
        int     time_idx = pgstat_get_io_time_index(io_op);
        int     byte_idx = pgstat_get_io_byte_index(io_op);

        /*
         * Some combinations of BackendType and IOOp, of IOContext and
         * IOOp, and of IOObject and IOOp are not tracked. Set these
         * cells in the view NULL.
         */
        if (pgstat_tracks_io_op(bktype, io_obj, io_context, io_op))
        {
          PgStat_Counter count =
            bktype_stats->counts[io_obj][io_context][io_op];

          values[op_idx] = Int64GetDatum(count);
        }
        else
          nulls[op_idx] = true;

        if (!nulls[op_idx])
        {
          /* not every operation is timed */
          if (time_idx != IO_COL_INVALID)
          {
            PgStat_Counter time =
              bktype_stats->times[io_obj][io_context][io_op];

            values[time_idx] = Float8GetDatum(pg_stat_us_to_ms(time));
          }

          /* not every IO is tracked in bytes */
          if (byte_idx != IO_COL_INVALID)
          {
            char    buf[256];
            PgStat_Counter byte =
              bktype_stats->bytes[io_obj][io_context][io_op];

            /* Convert to numeric */
            snprintf(buf, sizeof buf, INT64_FORMAT, byte);
            values[byte_idx] = DirectFunctionCall3(numeric_in,
                                 CStringGetDatum(buf),
                                 ObjectIdGetDatum(0),
                                 Int32GetDatum(-1));
          }
        }
        else
        {
          if (time_idx != IO_COL_INVALID)
            nulls[time_idx] = true;
          if (byte_idx != IO_COL_INVALID)
            nulls[byte_idx] = true;
        }
      }

      tuplestore_putvalues(rsinfo->setResult, rsinfo->setDesc,
                 values, nulls);
    }
  }
}

Datum
pg_stat_get_io(PG_FUNCTION_ARGS)
{
  ReturnSetInfo *rsinfo;
  PgStat_IO  *backends_io_stats;

  InitMaterializedSRF(fcinfo, 0);
  rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;

  backends_io_stats = pgstat_fetch_stat_io();

  for (int bktype = 0; bktype < BACKEND_NUM_TYPES; bktype++)
  {
    PgStat_BktypeIO *bktype_stats = &backends_io_stats->stats[bktype];

    /*
     * In Assert builds, we can afford an extra loop through all of the
     * counters (in pg_stat_io_build_tuples()), checking that only
     * expected stats are non-zero, since it keeps the non-Assert code
     * cleaner.
     */
    Assert(pgstat_bktype_io_stats_valid(bktype_stats, bktype));

    /*
     * For those BackendTypes without IO Operation stats, skip
     * representing them in the view altogether.
     */
    if (!pgstat_tracks_io_bktype(bktype))
      continue;

    /* save tuples with data from this PgStat_BktypeIO */
    pg_stat_io_build_tuples(rsinfo, bktype_stats, bktype,
                backends_io_stats->stat_reset_timestamp);
  }

  return (Datum) 0;
}

/*
 * Returns I/O statistics for a backend with given PID.
 */
Datum
pg_stat_get_backend_io(PG_FUNCTION_ARGS)
{
  ReturnSetInfo *rsinfo;
  BackendType bktype;
  int     pid;
  PgStat_Backend *backend_stats;
  PgStat_BktypeIO *bktype_stats;

  InitMaterializedSRF(fcinfo, 0);
  rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;

  pid = PG_GETARG_INT32(0);
  backend_stats = pgstat_fetch_stat_backend_by_pid(pid, &bktype);

  if (!backend_stats)
    return (Datum) 0;

  bktype_stats = &backend_stats->io_stats;

  /*
   * In Assert builds, we can afford an extra loop through all of the
   * counters (in pg_stat_io_build_tuples()), checking that only expected
   * stats are non-zero, since it keeps the non-Assert code cleaner.
   */
  Assert(pgstat_bktype_io_stats_valid(bktype_stats, bktype));

  /* save tuples with data from this PgStat_BktypeIO */
  pg_stat_io_build_tuples(rsinfo, bktype_stats, bktype,
              backend_stats->stat_reset_timestamp);
  return (Datum) 0;
}

/*
 * pg_stat_wal_build_tuple
 *
 * Helper routine for pg_stat_get_wal() and pg_stat_get_backend_wal()
 * returning one tuple based on the contents of wal_counters.
 */
static Datum
pg_stat_wal_build_tuple(PgStat_WalCounters wal_counters,
            TimestampTz stat_reset_timestamp)
{
#define PG_STAT_WAL_COLS  5
  TupleDesc tupdesc;
  Datum   values[PG_STAT_WAL_COLS] = {0};
  bool    nulls[PG_STAT_WAL_COLS] = {0};
  char    buf[256];

  /* Initialise attributes information in the tuple descriptor */
  tupdesc = CreateTemplateTupleDesc(PG_STAT_WAL_COLS);
  TupleDescInitEntry(tupdesc, (AttrNumber) 1, "wal_records",
             INT8OID, -1, 0);
  TupleDescInitEntry(tupdesc, (AttrNumber) 2, "wal_fpi",
             INT8OID, -1, 0);
  TupleDescInitEntry(tupdesc, (AttrNumber) 3, "wal_bytes",
             NUMERICOID, -1, 0);
  TupleDescInitEntry(tupdesc, (AttrNumber) 4, "wal_buffers_full",
             INT8OID, -1, 0);
  TupleDescInitEntry(tupdesc, (AttrNumber) 5, "stats_reset",
             TIMESTAMPTZOID, -1, 0);

  BlessTupleDesc(tupdesc);

  /* Fill values and NULLs */
  values[0] = Int64GetDatum(wal_counters.wal_records);
  values[1] = Int64GetDatum(wal_counters.wal_fpi);

  /* Convert to numeric. */
  snprintf(buf, sizeof buf, UINT64_FORMAT, wal_counters.wal_bytes);
  values[2] = DirectFunctionCall3(numeric_in,
                  CStringGetDatum(buf),
                  ObjectIdGetDatum(0),
                  Int32GetDatum(-1));

  values[3] = Int64GetDatum(wal_counters.wal_buffers_full);

  if (stat_reset_timestamp != 0)
    values[4] = TimestampTzGetDatum(stat_reset_timestamp);
  else
    nulls[4] = true;

  /* Returns the record as Datum */
  PG_RETURN_DATUM(HeapTupleGetDatum(heap_form_tuple(tupdesc, values, nulls)));
}

/*
 * Returns WAL statistics for a backend with given PID.
 */
Datum
pg_stat_get_backend_wal(PG_FUNCTION_ARGS)
{
  int     pid;
  PgStat_Backend *backend_stats;
  PgStat_WalCounters bktype_stats;

  pid = PG_GETARG_INT32(0);
  backend_stats = pgstat_fetch_stat_backend_by_pid(pid, NULL);

  if (!backend_stats)
    PG_RETURN_NULL();

  bktype_stats = backend_stats->wal_counters;

  /* save tuples with data from this PgStat_WalCounters */
  return (pg_stat_wal_build_tuple(bktype_stats, backend_stats->stat_reset_timestamp));
}

/*
 * Returns statistics of WAL activity
 */
Datum
pg_stat_get_wal(PG_FUNCTION_ARGS)
{
  PgStat_WalStats *wal_stats;

  /* Get statistics about WAL activity */
  wal_stats = pgstat_fetch_stat_wal();

  return (pg_stat_wal_build_tuple(wal_stats->wal_counters,
                  wal_stats->stat_reset_timestamp));
}

/*
 * Returns statistics of SLRU caches.
 */
Datum
pg_stat_get_slru(PG_FUNCTION_ARGS)
{
#define PG_STAT_GET_SLRU_COLS 9
  ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
  int     i;
  PgStat_SLRUStats *stats;

  InitMaterializedSRF(fcinfo, 0);

  /* request SLRU stats from the cumulative stats system */
  stats = pgstat_fetch_slru();

  for (i = 0;; i++)
  {
    /* for each row */
    Datum   values[PG_STAT_GET_SLRU_COLS] = {0};
    bool    nulls[PG_STAT_GET_SLRU_COLS] = {0};
    PgStat_SLRUStats stat;
    const char *name;

    name = pgstat_get_slru_name(i);

    if (!name)
      break;

    stat = stats[i];

    values[0] = PointerGetDatum(cstring_to_text(name));
    values[1] = Int64GetDatum(stat.blocks_zeroed);
    values[2] = Int64GetDatum(stat.blocks_hit);
    values[3] = Int64GetDatum(stat.blocks_read);
    values[4] = Int64GetDatum(stat.blocks_written);
    values[5] = Int64GetDatum(stat.blocks_exists);
    values[6] = Int64GetDatum(stat.flush);
    values[7] = Int64GetDatum(stat.truncate);
    values[8] = TimestampTzGetDatum(stat.stat_reset_timestamp);

    tuplestore_putvalues(rsinfo->setResult, rsinfo->setDesc, values, nulls);
  }

  return (Datum) 0;
}

#define PG_STAT_GET_XACT_RELENTRY_INT64(stat)     \
Datum                         \
CppConcat(pg_stat_get_xact_,stat)(PG_FUNCTION_ARGS)   \
{                           \
  Oid         relid = PG_GETARG_OID(0);        \
  int64       result;                 \
  PgStat_TableStatus *tabentry;           \
                            \
  if ((tabentry = find_tabstat_entry(relid)) == NULL) \
    result = 0;                   \
  else                        \
    result = (int64) (tabentry->counts.stat);   \
                            \
  PG_RETURN_INT64(result);             \
}

/* pg_stat_get_xact_numscans */
PG_STAT_GET_XACT_RELENTRY_INT64(numscans)

/* pg_stat_get_xact_tuples_returned */
PG_STAT_GET_XACT_RELENTRY_INT64(tuples_returned)

/* pg_stat_get_xact_tuples_fetched */
PG_STAT_GET_XACT_RELENTRY_INT64(tuples_fetched)

/* pg_stat_get_xact_tuples_hot_updated */
PG_STAT_GET_XACT_RELENTRY_INT64(tuples_hot_updated)

/* pg_stat_get_xact_tuples_newpage_updated */
PG_STAT_GET_XACT_RELENTRY_INT64(tuples_newpage_updated)

/* pg_stat_get_xact_blocks_fetched */
PG_STAT_GET_XACT_RELENTRY_INT64(blocks_fetched)

/* pg_stat_get_xact_blocks_hit */
PG_STAT_GET_XACT_RELENTRY_INT64(blocks_hit)

/* pg_stat_get_xact_tuples_inserted */
PG_STAT_GET_XACT_RELENTRY_INT64(tuples_inserted)

/* pg_stat_get_xact_tuples_updated */
PG_STAT_GET_XACT_RELENTRY_INT64(tuples_updated)

/* pg_stat_get_xact_tuples_deleted */
PG_STAT_GET_XACT_RELENTRY_INT64(tuples_deleted)

Datum
pg_stat_get_xact_function_calls(PG_FUNCTION_ARGS)
{
  Oid     funcid = PG_GETARG_OID(0);
  PgStat_FunctionCounts *funcentry;

  if ((funcentry = find_funcstat_entry(funcid)) == NULL)
    PG_RETURN_NULL();
  PG_RETURN_INT64(funcentry->numcalls);
}

#define PG_STAT_GET_XACT_FUNCENTRY_FLOAT8_MS(stat)        \
Datum                             \
CppConcat(pg_stat_get_xact_function_,stat)(PG_FUNCTION_ARGS)  \
{                               \
  Oid     funcid = PG_GETARG_OID(0);            \
  PgStat_FunctionCounts *funcentry;             \
                                \
  if ((funcentry = find_funcstat_entry(funcid)) == NULL)   \
    PG_RETURN_NULL();                   \
  PG_RETURN_FLOAT8(INSTR_TIME_GET_MILLISEC(funcentry->stat)); \
}

/* pg_stat_get_xact_function_total_time */
PG_STAT_GET_XACT_FUNCENTRY_FLOAT8_MS(total_time)

/* pg_stat_get_xact_function_self_time */
PG_STAT_GET_XACT_FUNCENTRY_FLOAT8_MS(self_time)

/* Get the timestamp of the current statistics snapshot */
Datum
pg_stat_get_snapshot_timestamp(PG_FUNCTION_ARGS)
{
  bool    have_snapshot;
  TimestampTz ts;

  ts = pgstat_get_stat_snapshot_timestamp(&have_snapshot);

  if (!have_snapshot)
    PG_RETURN_NULL();

  PG_RETURN_TIMESTAMPTZ(ts);
}

/* Discard the active statistics snapshot */
Datum
pg_stat_clear_snapshot(PG_FUNCTION_ARGS)
{
  pgstat_clear_snapshot();

  PG_RETURN_VOID();
}


/* Force statistics to be reported at the next occasion */
Datum
pg_stat_force_next_flush(PG_FUNCTION_ARGS)
{
  pgstat_force_next_flush();

  PG_RETURN_VOID();
}


/* Reset all counters for the current database */
Datum
pg_stat_reset(PG_FUNCTION_ARGS)
{
  pgstat_reset_counters();

  PG_RETURN_VOID();
}

/*
 * Reset some shared cluster-wide counters
 *
 * When adding a new reset target, ideally the name should match that in
 * pgstat_kind_builtin_infos, if relevant.
 */
Datum
pg_stat_reset_shared(PG_FUNCTION_ARGS)
{
  char     *target = NULL;

  if (PG_ARGISNULL(0))
  {
    /* Reset all the statistics when nothing is specified */
    pgstat_reset_of_kind(PGSTAT_KIND_ARCHIVER);
    pgstat_reset_of_kind(PGSTAT_KIND_BGWRITER);
    pgstat_reset_of_kind(PGSTAT_KIND_CHECKPOINTER);
    pgstat_reset_of_kind(PGSTAT_KIND_IO);
    XLogPrefetchResetStats();
    pgstat_reset_of_kind(PGSTAT_KIND_SLRU);
    pgstat_reset_of_kind(PGSTAT_KIND_WAL);

    PG_RETURN_VOID();
  }

  target = text_to_cstring(PG_GETARG_TEXT_PP(0));

  if (strcmp(target, "archiver") == 0)
    pgstat_reset_of_kind(PGSTAT_KIND_ARCHIVER);
  else if (strcmp(target, "bgwriter") == 0)
    pgstat_reset_of_kind(PGSTAT_KIND_BGWRITER);
  else if (strcmp(target, "checkpointer") == 0)
    pgstat_reset_of_kind(PGSTAT_KIND_CHECKPOINTER);
  else if (strcmp(target, "io") == 0)
    pgstat_reset_of_kind(PGSTAT_KIND_IO);
  else if (strcmp(target, "recovery_prefetch") == 0)
    XLogPrefetchResetStats();
  else if (strcmp(target, "slru") == 0)
    pgstat_reset_of_kind(PGSTAT_KIND_SLRU);
  else if (strcmp(target, "wal") == 0)
    pgstat_reset_of_kind(PGSTAT_KIND_WAL);
  else
    ereport(ERROR,
        (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
         errmsg("unrecognized reset target: \"%s\"", target),
         errhint("Target must be \"archiver\", \"bgwriter\", \"checkpointer\", \"io\", \"recovery_prefetch\", \"slru\", or \"wal\".")));

  PG_RETURN_VOID();
}

/*
 * Reset a statistics for a single object, which may be of current
 * database or shared across all databases in the cluster.
 */
Datum
pg_stat_reset_single_table_counters(PG_FUNCTION_ARGS)
{
  Oid     taboid = PG_GETARG_OID(0);
  Oid     dboid = (IsSharedRelation(taboid) ? InvalidOid : MyDatabaseId);

  pgstat_reset(PGSTAT_KIND_RELATION, dboid, taboid);

  PG_RETURN_VOID();
}

Datum
pg_stat_reset_single_function_counters(PG_FUNCTION_ARGS)
{
  Oid     funcoid = PG_GETARG_OID(0);

  pgstat_reset(PGSTAT_KIND_FUNCTION, MyDatabaseId, funcoid);

  PG_RETURN_VOID();
}

/*
 * Reset statistics of backend with given PID.
 */
Datum
pg_stat_reset_backend_stats(PG_FUNCTION_ARGS)
{
  PGPROC     *proc;
  PgBackendStatus *beentry;
  ProcNumber  procNumber;
  int     backend_pid = PG_GETARG_INT32(0);

  proc = BackendPidGetProc(backend_pid);

  /* This could be an auxiliary process */
  if (!proc)
    proc = AuxiliaryPidGetProc(backend_pid);

  if (!proc)
    PG_RETURN_VOID();

  procNumber = GetNumberFromPGProc(proc);

  beentry = pgstat_get_beentry_by_proc_number(procNumber);
  if (!beentry)
    PG_RETURN_VOID();

  /* Check if the backend type tracks statistics */
  if (!pgstat_tracks_backend_bktype(beentry->st_backendType))
    PG_RETURN_VOID();

  pgstat_reset(PGSTAT_KIND_BACKEND, InvalidOid, procNumber);

  PG_RETURN_VOID();
}

/* Reset SLRU counters (a specific one or all of them). */
Datum
pg_stat_reset_slru(PG_FUNCTION_ARGS)
{
  char     *target = NULL;

  if (PG_ARGISNULL(0))
    pgstat_reset_of_kind(PGSTAT_KIND_SLRU);
  else
  {
    target = text_to_cstring(PG_GETARG_TEXT_PP(0));
    pgstat_reset_slru(target);
  }

  PG_RETURN_VOID();
}

/* Reset replication slots stats (a specific one or all of them). */
Datum
pg_stat_reset_replication_slot(PG_FUNCTION_ARGS)
{
  char     *target = NULL;

  if (PG_ARGISNULL(0))
    pgstat_reset_of_kind(PGSTAT_KIND_REPLSLOT);
  else
  {
    target = text_to_cstring(PG_GETARG_TEXT_PP(0));
    pgstat_reset_replslot(target);
  }

  PG_RETURN_VOID();
}

/* Reset subscription stats (a specific one or all of them) */
Datum
pg_stat_reset_subscription_stats(PG_FUNCTION_ARGS)
{
  Oid     subid;

  if (PG_ARGISNULL(0))
  {
    /* Clear all subscription stats */
    pgstat_reset_of_kind(PGSTAT_KIND_SUBSCRIPTION);
  }
  else
  {
    subid = PG_GETARG_OID(0);

    if (!OidIsValid(subid))
      ereport(ERROR,
          (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
           errmsg("invalid subscription OID %u", subid)));
    pgstat_reset(PGSTAT_KIND_SUBSCRIPTION, InvalidOid, subid);
  }

  PG_RETURN_VOID();
}

Datum
pg_stat_get_archiver(PG_FUNCTION_ARGS)
{
  TupleDesc tupdesc;
  Datum   values[7] = {0};
  bool    nulls[7] = {0};
  PgStat_ArchiverStats *archiver_stats;

  /* Initialise attributes information in the tuple descriptor */
  tupdesc = CreateTemplateTupleDesc(7);
  TupleDescInitEntry(tupdesc, (AttrNumber) 1, "archived_count",
             INT8OID, -1, 0);
  TupleDescInitEntry(tupdesc, (AttrNumber) 2, "last_archived_wal",
             TEXTOID, -1, 0);
  TupleDescInitEntry(tupdesc, (AttrNumber) 3, "last_archived_time",
             TIMESTAMPTZOID, -1, 0);
  TupleDescInitEntry(tupdesc, (AttrNumber) 4, "failed_count",
             INT8OID, -1, 0);
  TupleDescInitEntry(tupdesc, (AttrNumber) 5, "last_failed_wal",
             TEXTOID, -1, 0);
  TupleDescInitEntry(tupdesc, (AttrNumber) 6, "last_failed_time",
             TIMESTAMPTZOID, -1, 0);
  TupleDescInitEntry(tupdesc, (AttrNumber) 7, "stats_reset",
             TIMESTAMPTZOID, -1, 0);

  BlessTupleDesc(tupdesc);

  /* Get statistics about the archiver process */
  archiver_stats = pgstat_fetch_stat_archiver();

  /* Fill values and NULLs */
  values[0] = Int64GetDatum(archiver_stats->archived_count);
  if (*(archiver_stats->last_archived_wal) == '\0')
    nulls[1] = true;
  else
    values[1] = CStringGetTextDatum(archiver_stats->last_archived_wal);

  if (archiver_stats->last_archived_timestamp == 0)
    nulls[2] = true;
  else
    values[2] = TimestampTzGetDatum(archiver_stats->last_archived_timestamp);

  values[3] = Int64GetDatum(archiver_stats->failed_count);
  if (*(archiver_stats->last_failed_wal) == '\0')
    nulls[4] = true;
  else
    values[4] = CStringGetTextDatum(archiver_stats->last_failed_wal);

  if (archiver_stats->last_failed_timestamp == 0)
    nulls[5] = true;
  else
    values[5] = TimestampTzGetDatum(archiver_stats->last_failed_timestamp);

  if (archiver_stats->stat_reset_timestamp == 0)
    nulls[6] = true;
  else
    values[6] = TimestampTzGetDatum(archiver_stats->stat_reset_timestamp);

  /* Returns the record as Datum */
  PG_RETURN_DATUM(HeapTupleGetDatum(heap_form_tuple(tupdesc, values, nulls)));
}

/*
 * Get the statistics for the replication slot. If the slot statistics is not
 * available, return all-zeroes stats.
 */
Datum
pg_stat_get_replication_slot(PG_FUNCTION_ARGS)
{
#define PG_STAT_GET_REPLICATION_SLOT_COLS 10
  text     *slotname_text = PG_GETARG_TEXT_P(0);
  NameData  slotname;
  TupleDesc tupdesc;
  Datum   values[PG_STAT_GET_REPLICATION_SLOT_COLS] = {0};
  bool    nulls[PG_STAT_GET_REPLICATION_SLOT_COLS] = {0};
  PgStat_StatReplSlotEntry *slotent;
  PgStat_StatReplSlotEntry allzero;

  /* Initialise attributes information in the tuple descriptor */
  tupdesc = CreateTemplateTupleDesc(PG_STAT_GET_REPLICATION_SLOT_COLS);
  TupleDescInitEntry(tupdesc, (AttrNumber) 1, "slot_name",
             TEXTOID, -1, 0);
  TupleDescInitEntry(tupdesc, (AttrNumber) 2, "spill_txns",
             INT8OID, -1, 0);
  TupleDescInitEntry(tupdesc, (AttrNumber) 3, "spill_count",
             INT8OID, -1, 0);
  TupleDescInitEntry(tupdesc, (AttrNumber) 4, "spill_bytes",
             INT8OID, -1, 0);
  TupleDescInitEntry(tupdesc, (AttrNumber) 5, "stream_txns",
             INT8OID, -1, 0);
  TupleDescInitEntry(tupdesc, (AttrNumber) 6, "stream_count",
             INT8OID, -1, 0);
  TupleDescInitEntry(tupdesc, (AttrNumber) 7, "stream_bytes",
             INT8OID, -1, 0);
  TupleDescInitEntry(tupdesc, (AttrNumber) 8, "total_txns",
             INT8OID, -1, 0);
  TupleDescInitEntry(tupdesc, (AttrNumber) 9, "total_bytes",
             INT8OID, -1, 0);
  TupleDescInitEntry(tupdesc, (AttrNumber) 10, "stats_reset",
             TIMESTAMPTZOID, -1, 0);
  BlessTupleDesc(tupdesc);

  namestrcpy(&slotname, text_to_cstring(slotname_text));
  slotent = pgstat_fetch_replslot(slotname);
  if (!slotent)
  {
    /*
     * If the slot is not found, initialise its stats. This is possible if
     * the create slot message is lost.
     */
    memset(&allzero, 0, sizeof(PgStat_StatReplSlotEntry));
    slotent = &allzero;
  }

  values[0] = CStringGetTextDatum(NameStr(slotname));
  values[1] = Int64GetDatum(slotent->spill_txns);
  values[2] = Int64GetDatum(slotent->spill_count);
  values[3] = Int64GetDatum(slotent->spill_bytes);
  values[4] = Int64GetDatum(slotent->stream_txns);
  values[5] = Int64GetDatum(slotent->stream_count);
  values[6] = Int64GetDatum(slotent->stream_bytes);
  values[7] = Int64GetDatum(slotent->total_txns);
  values[8] = Int64GetDatum(slotent->total_bytes);

  if (slotent->stat_reset_timestamp == 0)
    nulls[9] = true;
  else
    values[9] = TimestampTzGetDatum(slotent->stat_reset_timestamp);

  /* Returns the record as Datum */
  PG_RETURN_DATUM(HeapTupleGetDatum(heap_form_tuple(tupdesc, values, nulls)));
}

/*
 * Get the subscription statistics for the given subscription. If the
 * subscription statistics is not available, return all-zeros stats.
 */
Datum
pg_stat_get_subscription_stats(PG_FUNCTION_ARGS)
{
#define PG_STAT_GET_SUBSCRIPTION_STATS_COLS 11
  Oid     subid = PG_GETARG_OID(0);
  TupleDesc tupdesc;
  Datum   values[PG_STAT_GET_SUBSCRIPTION_STATS_COLS] = {0};
  bool    nulls[PG_STAT_GET_SUBSCRIPTION_STATS_COLS] = {0};
  PgStat_StatSubEntry *subentry;
  PgStat_StatSubEntry allzero;
  int     i = 0;

  /* Get subscription stats */
  subentry = pgstat_fetch_stat_subscription(subid);

  /* Initialise attributes information in the tuple descriptor */
  tupdesc = CreateTemplateTupleDesc(PG_STAT_GET_SUBSCRIPTION_STATS_COLS);
  TupleDescInitEntry(tupdesc, (AttrNumber) 1, "subid",
             OIDOID, -1, 0);
  TupleDescInitEntry(tupdesc, (AttrNumber) 2, "apply_error_count",
             INT8OID, -1, 0);
  TupleDescInitEntry(tupdesc, (AttrNumber) 3, "sync_error_count",
             INT8OID, -1, 0);
  TupleDescInitEntry(tupdesc, (AttrNumber) 4, "confl_insert_exists",
             INT8OID, -1, 0);
  TupleDescInitEntry(tupdesc, (AttrNumber) 5, "confl_update_origin_differs",
             INT8OID, -1, 0);
  TupleDescInitEntry(tupdesc, (AttrNumber) 6, "confl_update_exists",
             INT8OID, -1, 0);
  TupleDescInitEntry(tupdesc, (AttrNumber) 7, "confl_update_missing",
             INT8OID, -1, 0);
  TupleDescInitEntry(tupdesc, (AttrNumber) 8, "confl_delete_origin_differs",
             INT8OID, -1, 0);
  TupleDescInitEntry(tupdesc, (AttrNumber) 9, "confl_delete_missing",
             INT8OID, -1, 0);
  TupleDescInitEntry(tupdesc, (AttrNumber) 10, "confl_multiple_unique_conflicts",
             INT8OID, -1, 0);
  TupleDescInitEntry(tupdesc, (AttrNumber) 11, "stats_reset",
             TIMESTAMPTZOID, -1, 0);
  BlessTupleDesc(tupdesc);

  if (!subentry)
  {
    /* If the subscription is not found, initialise its stats */
    memset(&allzero, 0, sizeof(PgStat_StatSubEntry));
    subentry = &allzero;
  }

  /* subid */
  values[i++] = ObjectIdGetDatum(subid);

  /* apply_error_count */
  values[i++] = Int64GetDatum(subentry->apply_error_count);

  /* sync_error_count */
  values[i++] = Int64GetDatum(subentry->sync_error_count);

  /* conflict count */
  for (int nconflict = 0; nconflict < CONFLICT_NUM_TYPES; nconflict++)
    values[i++] = Int64GetDatum(subentry->conflict_count[nconflict]);

  /* stats_reset */
  if (subentry->stat_reset_timestamp == 0)
    nulls[i] = true;
  else
    values[i] = TimestampTzGetDatum(subentry->stat_reset_timestamp);

  Assert(i + 1 == PG_STAT_GET_SUBSCRIPTION_STATS_COLS);

  /* Returns the record as Datum */
  PG_RETURN_DATUM(HeapTupleGetDatum(heap_form_tuple(tupdesc, values, nulls)));
}

/*
 * Checks for presence of stats for object with provided kind, database oid,
 * object oid.
 *
 * This is useful for tests, but not really anything else. Therefore not
 * documented.
 */
Datum
pg_stat_have_stats(PG_FUNCTION_ARGS)
{
  char     *stats_type = text_to_cstring(PG_GETARG_TEXT_P(0));
  Oid     dboid = PG_GETARG_OID(1);
  uint64    objid = PG_GETARG_INT64(2);
  PgStat_Kind kind = pgstat_get_kind_from_str(stats_type);

  PG_RETURN_BOOL(pgstat_have_entry(kind, dboid, objid));
}

Coverage Report

Created: 2025-07-03 06:49