/*----------------------------------------------------------------------

 *

 * tableam.c

 *    Table access method routines too big to be inline functions.

 *

 * Portions Copyright (c) 1996-2026, PostgreSQL Global Development Group

 * Portions Copyright (c) 1994, Regents of the University of California

 *

 *

 * IDENTIFICATION

 *    src/backend/access/table/tableam.c

 *

 * NOTES

 *    Note that most functions in here are documented in tableam.h, rather than

 *    here. That's because there's a lot of inline functions in tableam.h and

 *    it'd be harder to understand if one constantly had to switch between files.

 *

 *----------------------------------------------------------------------

 */

#include "postgres.h"

#include <math.h>

#include "access/syncscan.h"

#include "access/tableam.h"

#include "access/xact.h"

#include "optimizer/optimizer.h"

#include "optimizer/plancat.h"

#include "port/pg_bitutils.h"

#include "storage/bufmgr.h"

#include "storage/shmem.h"

#include "storage/smgr.h"

/*

 * Constants to control the behavior of block allocation to parallel workers

 * during a parallel seqscan.  Technically these values do not need to be

 * powers of 2, but having them as powers of 2 makes the math more optimal

 * and makes the ramp-down stepping more even.

 */

/* The number of I/O chunks we try to break a parallel seqscan down into */

#define PARALLEL_SEQSCAN_NCHUNKS      2048

/* Ramp down size of allocations when we've only this number of chunks left */

#define PARALLEL_SEQSCAN_RAMPDOWN_CHUNKS  64

/* Cap the size of parallel I/O chunks to this number of blocks */

#define PARALLEL_SEQSCAN_MAX_CHUNK_SIZE   8192

/* GUC variables */

char     *default_table_access_method = DEFAULT_TABLE_ACCESS_METHOD;

bool    synchronize_seqscans = true;

/* ----------------------------------------------------------------------------

 * Slot functions.

 * ----------------------------------------------------------------------------

 */

const TupleTableSlotOps *

table_slot_callbacks(Relation relation)

{

  const TupleTableSlotOps *tts_cb;

  if (relation->rd_tableam)

    tts_cb = relation->rd_tableam->slot_callbacks(relation);

  else if (relation->rd_rel->relkind == RELKIND_FOREIGN_TABLE)

  {

    /*

     * Historically FDWs expect to store heap tuples in slots. Continue

     * handing them one, to make it less painful to adapt FDWs to new

     * versions. The cost of a heap slot over a virtual slot is pretty

     * small.

     */

    tts_cb = &TTSOpsHeapTuple;

  }

  else

  {

    /*

     * These need to be supported, as some parts of the code (like COPY)

     * need to create slots for such relations too. It seems better to

     * centralize the knowledge that a heap slot is the right thing in

     * that case here.

     */

    Assert(relation->rd_rel->relkind == RELKIND_VIEW ||

         relation->rd_rel->relkind == RELKIND_PARTITIONED_TABLE);

    tts_cb = &TTSOpsVirtual;

  }

  return tts_cb;

}

TupleTableSlot *

table_slot_create(Relation relation, List **reglist)

{

  const TupleTableSlotOps *tts_cb;

  TupleTableSlot *slot;

  tts_cb = table_slot_callbacks(relation);

  slot = MakeSingleTupleTableSlot(RelationGetDescr(relation), tts_cb);

  if (reglist)

    *reglist = lappend(*reglist, slot);

  return slot;

}

/* ----------------------------------------------------------------------------

 * Table scan functions.

 * ----------------------------------------------------------------------------

 */

TableScanDesc

table_beginscan_catalog(Relation relation, int nkeys, ScanKeyData *key)

{

  uint32    flags = SO_TYPE_SEQSCAN |

    SO_ALLOW_STRAT | SO_ALLOW_SYNC | SO_ALLOW_PAGEMODE | SO_TEMP_SNAPSHOT;

  Oid     relid = RelationGetRelid(relation);

  Snapshot  snapshot = RegisterSnapshot(GetCatalogSnapshot(relid));

  return table_beginscan_common(relation, snapshot, nkeys, key,

                  NULL, flags, SO_NONE);

}

/* ----------------------------------------------------------------------------

 * Parallel table scan related functions.

 * ----------------------------------------------------------------------------

 */

Size

table_parallelscan_estimate(Relation rel, Snapshot snapshot)

{

  Size    sz = 0;

  if (IsMVCCSnapshot(snapshot))

    sz = add_size(sz, EstimateSnapshotSpace(snapshot));

  else

    Assert(snapshot == SnapshotAny);

  sz = add_size(sz, rel->rd_tableam->parallelscan_estimate(rel));

  return sz;

}

void

table_parallelscan_initialize(Relation rel, ParallelTableScanDesc pscan,

                Snapshot snapshot)

{

  Size    snapshot_off = rel->rd_tableam->parallelscan_initialize(rel, pscan);

  pscan->phs_snapshot_off = snapshot_off;

  if (IsMVCCSnapshot(snapshot))

  {

    SerializeSnapshot(snapshot, (char *) pscan + pscan->phs_snapshot_off);

    pscan->phs_snapshot_any = false;

  }

  else

  {

    Assert(snapshot == SnapshotAny);

    pscan->phs_snapshot_any = true;

  }

}

TableScanDesc

table_beginscan_parallel(Relation relation, ParallelTableScanDesc pscan,

             uint32 flags)

{

  Snapshot  snapshot;

  uint32    internal_flags = SO_TYPE_SEQSCAN |

    SO_ALLOW_STRAT | SO_ALLOW_SYNC | SO_ALLOW_PAGEMODE;

  Assert(RelFileLocatorEquals(relation->rd_locator, pscan->phs_locator));

  if (!pscan->phs_snapshot_any)

  {

    /* Snapshot was serialized -- restore it */

    snapshot = RestoreSnapshot((char *) pscan + pscan->phs_snapshot_off);

    RegisterSnapshot(snapshot);

    internal_flags |= SO_TEMP_SNAPSHOT;

  }

  else

  {

    /* SnapshotAny passed by caller (not serialized) */

    snapshot = SnapshotAny;

  }

  return table_beginscan_common(relation, snapshot, 0, NULL,

                  pscan, internal_flags, flags);

}

TableScanDesc

table_beginscan_parallel_tidrange(Relation relation,

                  ParallelTableScanDesc pscan,

                  uint32 flags)

{

  Snapshot  snapshot;

  TableScanDesc sscan;

  uint32    internal_flags = SO_TYPE_TIDRANGESCAN | SO_ALLOW_PAGEMODE;

  Assert(RelFileLocatorEquals(relation->rd_locator, pscan->phs_locator));

  /* disable syncscan in parallel tid range scan. */

  pscan->phs_syncscan = false;

  if (!pscan->phs_snapshot_any)

  {

    /* Snapshot was serialized -- restore it */

    snapshot = RestoreSnapshot((char *) pscan + pscan->phs_snapshot_off);

    RegisterSnapshot(snapshot);

    internal_flags |= SO_TEMP_SNAPSHOT;

  }

  else

  {

    /* SnapshotAny passed by caller (not serialized) */

    snapshot = SnapshotAny;

  }

  sscan = table_beginscan_common(relation, snapshot, 0, NULL,

                   pscan, internal_flags, flags);

  return sscan;

}

/* ------------------------------------------------------------------------

 * Functions for non-modifying operations on individual tuples

 * ------------------------------------------------------------------------

 */

void

table_tuple_get_latest_tid(TableScanDesc scan, ItemPointer tid)

{

  Relation  rel = scan->rs_rd;

  const TableAmRoutine *tableam = rel->rd_tableam;

  /*

   * Since this can be called with user-supplied TID, don't trust the input

   * too much.

   */

  if (!tableam->tuple_tid_valid(scan, tid))

    ereport(ERROR,

        (errcode(ERRCODE_INVALID_PARAMETER_VALUE),

         errmsg("tid (%u, %u) is not valid for relation \"%s\"",

            ItemPointerGetBlockNumberNoCheck(tid),

            ItemPointerGetOffsetNumberNoCheck(tid),

            RelationGetRelationName(rel))));

  tableam->tuple_get_latest_tid(scan, tid);

}

/* ----------------------------------------------------------------------------

 * Functions to make modifications a bit simpler.

 * ----------------------------------------------------------------------------

 */

/*

 * simple_table_tuple_insert - insert a tuple

 *

 * Currently, this routine differs from table_tuple_insert only in supplying a

 * default command ID and not allowing access to the speedup options.

 */

void

simple_table_tuple_insert(Relation rel, TupleTableSlot *slot)

{

  table_tuple_insert(rel, slot, GetCurrentCommandId(true), 0, NULL);

}

/*

 * simple_table_tuple_delete - delete a tuple

 *

 * This routine may be used to delete a tuple when concurrent updates of

 * the target tuple are not expected (for example, because we have a lock

 * on the relation associated with the tuple).  Any failure is reported

 * via ereport().

 */

void

simple_table_tuple_delete(Relation rel, ItemPointer tid, Snapshot snapshot)

{

  TM_Result result;

  TM_FailureData tmfd;

  result = table_tuple_delete(rel, tid,

                GetCurrentCommandId(true),

                0, snapshot, InvalidSnapshot,

                true /* wait for commit */ ,

                &tmfd);

  switch (result)

  {

    case TM_SelfModified:

      /* Tuple was already updated in current command? */

      elog(ERROR, "tuple already updated by self");

      break;

    case TM_Ok:

      /* done successfully */

      break;

    case TM_Updated:

      elog(ERROR, "tuple concurrently updated");

      break;

    case TM_Deleted:

      elog(ERROR, "tuple concurrently deleted");

      break;

    default:

      elog(ERROR, "unrecognized table_tuple_delete status: %u", result);

      break;

  }

}

/*

 * simple_table_tuple_update - replace a tuple

 *

 * This routine may be used to update a tuple when concurrent updates of

 * the target tuple are not expected (for example, because we have a lock

 * on the relation associated with the tuple).  Any failure is reported

 * via ereport().

 */

void

simple_table_tuple_update(Relation rel, ItemPointer otid,

              TupleTableSlot *slot,

              Snapshot snapshot,

              TU_UpdateIndexes *update_indexes)

{

  TM_Result result;

  TM_FailureData tmfd;

  LockTupleMode lockmode;

  result = table_tuple_update(rel, otid, slot,

                GetCurrentCommandId(true),

                0, snapshot, InvalidSnapshot,

                true /* wait for commit */ ,

                &tmfd, &lockmode, update_indexes);

  switch (result)

  {

    case TM_SelfModified:

      /* Tuple was already updated in current command? */

      elog(ERROR, "tuple already updated by self");

      break;

    case TM_Ok:

      /* done successfully */

      break;

    case TM_Updated:

      elog(ERROR, "tuple concurrently updated");

      break;

    case TM_Deleted:

      elog(ERROR, "tuple concurrently deleted");

      break;

    default:

      elog(ERROR, "unrecognized table_tuple_update status: %u", result);

      break;

  }

}

/* ----------------------------------------------------------------------------

 * Helper functions to implement parallel scans for block oriented AMs.

 * ----------------------------------------------------------------------------

 */

Size

table_block_parallelscan_estimate(Relation rel)

{

  return sizeof(ParallelBlockTableScanDescData);

}

Size

table_block_parallelscan_initialize(Relation rel, ParallelTableScanDesc pscan)

{

  ParallelBlockTableScanDesc bpscan = (ParallelBlockTableScanDesc) pscan;

  bpscan->base.phs_locator = rel->rd_locator;

  bpscan->phs_nblocks = RelationGetNumberOfBlocks(rel);

  /* compare phs_syncscan initialization to similar logic in initscan */

  bpscan->base.phs_syncscan = synchronize_seqscans &&

    !RelationUsesLocalBuffers(rel) &&

    bpscan->phs_nblocks > NBuffers / 4;

  pg_atomic_init_u32(&bpscan->phs_startblock, InvalidBlockNumber);

  pg_atomic_init_u32(&bpscan->phs_numblock, InvalidBlockNumber);

  pg_atomic_init_u64(&bpscan->phs_nallocated, 0);

  return sizeof(ParallelBlockTableScanDescData);

}

void

table_block_parallelscan_reinitialize(Relation rel, ParallelTableScanDesc pscan)

{

  ParallelBlockTableScanDesc bpscan = (ParallelBlockTableScanDesc) pscan;

  /*

   * Reset the start block and block limit too.  These are set by the first

   * participant to call table_block_parallelscan_startblock_init(), and a

   * rescan may want to scan a different range of blocks, e.g. a TID Range

   * Scan with different TID bounds.

   */

  pg_atomic_write_u32(&bpscan->phs_startblock, InvalidBlockNumber);

  pg_atomic_write_u32(&bpscan->phs_numblock, InvalidBlockNumber);

  pg_atomic_write_u64(&bpscan->phs_nallocated, 0);

}

/*

 * find and set the scan's startblock

 *

 * Determine where the parallel seq scan should start.  This function may be

 * called many times, once by each parallel worker.  We must be careful only

 * to set the phs_startblock and phs_numblock fields once.

 *

 * Callers may optionally specify a non-InvalidBlockNumber value for

 * 'startblock' to force the scan to start at the given page.  Likewise,

 * 'numblocks' can be specified as a non-InvalidBlockNumber to limit the

 * number of blocks to scan to that many blocks.

 */

void

table_block_parallelscan_startblock_init(Relation rel,

                     ParallelBlockTableScanWorker pbscanwork,

                     ParallelBlockTableScanDesc pbscan,

                     BlockNumber startblock,

                     BlockNumber numblocks)

{

  StaticAssertDecl(MaxBlockNumber <= 0xFFFFFFFE,

           "pg_nextpower2_32 may be too small for non-standard BlockNumber width");

  BlockNumber scan_nblocks;

  /* Reset the state we use for controlling allocation size. */

  memset(pbscanwork, 0, sizeof(*pbscanwork));

  /*

   * When the caller specified a limit on the number of blocks to scan, set

   * that in the ParallelBlockTableScanDesc, if it's not been done by

   * another worker already.

   */

  if (numblocks != InvalidBlockNumber)

  {

    uint32    expected = InvalidBlockNumber;

    pg_atomic_compare_exchange_u32(&pbscan->phs_numblock, &expected,

                     numblocks);

  }

  /*

   * If the scan's phs_startblock has not yet been initialized, we must do

   * so now.  If a startblock was specified, start there, otherwise if this

   * is not a synchronized scan, we just start at block 0, but if it is a

   * synchronized scan, we must get the starting position from the

   * synchronized scan machinery.

   *

   * If another worker initializes phs_startblock concurrently, just use

   * their value.

   */

  if (pg_atomic_read_u32(&pbscan->phs_startblock) == InvalidBlockNumber)

  {

    BlockNumber newstartblock;

    uint32    expected = InvalidBlockNumber;

    if (startblock != InvalidBlockNumber)

      newstartblock = startblock;

    else if (!pbscan->base.phs_syncscan)

      newstartblock = 0;

    else

      newstartblock = ss_get_location(rel, pbscan->phs_nblocks);

    pg_atomic_compare_exchange_u32(&pbscan->phs_startblock, &expected,

                     newstartblock);

  }

  /*

   * Figure out how many blocks we're going to scan; either all of them, or

   * just phs_numblock's worth, if a limit has been imposed.

   */

  if (pg_atomic_read_u32(&pbscan->phs_numblock) == InvalidBlockNumber)

    scan_nblocks = pbscan->phs_nblocks;

  else

    scan_nblocks = pg_atomic_read_u32(&pbscan->phs_numblock);

  /*

   * We determine the chunk size based on scan_nblocks.  First we split

   * scan_nblocks into PARALLEL_SEQSCAN_NCHUNKS chunks then we calculate the

   * next highest power of 2 number of the result.  This means we split the

   * blocks we're scanning into somewhere between PARALLEL_SEQSCAN_NCHUNKS

   * and PARALLEL_SEQSCAN_NCHUNKS / 2 chunks.

   */

  pbscanwork->phsw_chunk_size = pg_nextpower2_32(Max(scan_nblocks /

                             PARALLEL_SEQSCAN_NCHUNKS, 1));

  /*

   * Ensure we don't go over the maximum chunk size with larger tables. This

   * means we may get much more than PARALLEL_SEQSCAN_NCHUNKS for larger

   * tables.  Too large a chunk size has been shown to be detrimental to

   * sequential scan performance.

   */

  pbscanwork->phsw_chunk_size = Min(pbscanwork->phsw_chunk_size,

                    PARALLEL_SEQSCAN_MAX_CHUNK_SIZE);

}

/*

 * get the next page to scan

 *

 * Get the next page to scan.  Even if there are no pages left to scan,

 * another backend could have grabbed a page to scan and not yet finished

 * looking at it, so it doesn't follow that the scan is done when the first

 * backend gets an InvalidBlockNumber return.

 */

BlockNumber

table_block_parallelscan_nextpage(Relation rel,

                  ParallelBlockTableScanWorker pbscanwork,

                  ParallelBlockTableScanDesc pbscan)

{

  BlockNumber scan_nblocks;

  BlockNumber page;

  uint64    nallocated;

  /*

   * The logic below allocates block numbers out to parallel workers in a

   * way that each worker will receive a set of consecutive block numbers to

   * scan.  Earlier versions of this would allocate the next highest block

   * number to the next worker to call this function.  This would generally

   * result in workers never receiving consecutive block numbers.  Some

   * operating systems would not detect the sequential I/O pattern due to

   * each backend being a different process which could result in poor

   * performance due to inefficient or no readahead.  To work around this

   * issue, we now allocate a range of block numbers for each worker and

   * when they come back for another block, we give them the next one in

   * that range until the range is complete.  When the worker completes the

   * range of blocks we then allocate another range for it and return the

   * first block number from that range.

   *

   * Here we name these ranges of blocks "chunks".  The initial size of

   * these chunks is determined in table_block_parallelscan_startblock_init

   * based on the number of blocks to scan.  Towards the end of the scan, we

   * start making reductions in the size of the chunks in order to attempt

   * to divide the remaining work over all the workers as evenly as

   * possible.

   *

   * Here pbscanwork is local worker memory.  phsw_chunk_remaining tracks

   * the number of blocks remaining in the chunk.  When that reaches 0 then

   * we must allocate a new chunk for the worker.

   *

   * phs_nallocated tracks how many blocks have been allocated to workers

   * already.  When phs_nallocated >= rs_nblocks, all blocks have been

   * allocated.

   *

   * Because we use an atomic fetch-and-add to fetch the current value, the

   * phs_nallocated counter will exceed rs_nblocks, because workers will

   * still increment the value, when they try to allocate the next block but

   * all blocks have been allocated already. The counter must be 64 bits

   * wide because of that, to avoid wrapping around when scan_nblocks is

   * close to 2^32.

   *

   * The actual block to return is calculated by adding the counter to the

   * starting block number, modulo phs_nblocks.

   */

  /* First, figure out how many blocks we're planning on scanning */

  if (pg_atomic_read_u32(&pbscan->phs_numblock) == InvalidBlockNumber)

    scan_nblocks = pbscan->phs_nblocks;

  else

    scan_nblocks = pg_atomic_read_u32(&pbscan->phs_numblock);

  /*

   * Now check if we have any remaining blocks in a previous chunk for this

   * worker.  We must consume all of the blocks from that before we allocate

   * a new chunk to the worker.

   */

  if (pbscanwork->phsw_chunk_remaining > 0)

  {

    /*

     * Give them the next block in the range and update the remaining

     * number of blocks.

     */

    nallocated = ++pbscanwork->phsw_nallocated;

    pbscanwork->phsw_chunk_remaining--;

  }

  else

  {

    /*

     * When we've only got PARALLEL_SEQSCAN_RAMPDOWN_CHUNKS chunks

     * remaining in the scan, we half the chunk size.  Since we reduce the

     * chunk size here, we'll hit this again after doing

     * PARALLEL_SEQSCAN_RAMPDOWN_CHUNKS at the new size.  After a few

     * iterations of this, we'll end up doing the last few blocks with the

     * chunk size set to 1.

     */

    if (pbscanwork->phsw_chunk_size > 1 &&

      pbscanwork->phsw_nallocated > scan_nblocks -

      (pbscanwork->phsw_chunk_size * PARALLEL_SEQSCAN_RAMPDOWN_CHUNKS))

      pbscanwork->phsw_chunk_size >>= 1;

    nallocated = pbscanwork->phsw_nallocated =

      pg_atomic_fetch_add_u64(&pbscan->phs_nallocated,

                  pbscanwork->phsw_chunk_size);

    /*

     * Set the remaining number of blocks in this chunk so that subsequent

     * calls from this worker continue on with this chunk until it's done.

     */

    pbscanwork->phsw_chunk_remaining = pbscanwork->phsw_chunk_size - 1;

  }

  /* Check if we've run out of blocks to scan */

  if (nallocated >= scan_nblocks)

    page = InvalidBlockNumber; /* all blocks have been allocated */

  else

    page = (nallocated + pg_atomic_read_u32(&pbscan->phs_startblock)) % pbscan->phs_nblocks;

  /*

   * Report scan location.  Normally, we report the current page number.

   * When we reach the end of the scan, though, we report the starting page,

   * not the ending page, just so the starting positions for later scans

   * doesn't slew backwards.  We only report the position at the end of the

   * scan once, though: subsequent callers will report nothing.

   */

  if (pbscan->base.phs_syncscan)

  {

    if (page != InvalidBlockNumber)

      ss_report_location(rel, page);

    else if (nallocated == pbscan->phs_nblocks)

      ss_report_location(rel, pg_atomic_read_u32(&pbscan->phs_startblock));

  }

  return page;

}

/* ----------------------------------------------------------------------------

 * Helper functions to implement relation sizing for block oriented AMs.

 * ----------------------------------------------------------------------------

 */

/*

 * table_block_relation_size

 *

 * If a table AM uses the various relation forks as the sole place where data

 * is stored, and if it uses them in the expected manner (e.g. the actual data

 * is in the main fork rather than some other), it can use this implementation

 * of the relation_size callback rather than implementing its own.

 */

uint64

table_block_relation_size(Relation rel, ForkNumber forkNumber)

{

  uint64    nblocks = 0;

  /* InvalidForkNumber indicates returning the size for all forks */

  if (forkNumber == InvalidForkNumber)

  {

    for (int i = 0; i < MAX_FORKNUM; i++)

      nblocks += smgrnblocks(RelationGetSmgr(rel), i);

  }

  else

    nblocks = smgrnblocks(RelationGetSmgr(rel), forkNumber);

  return nblocks * BLCKSZ;

}

/*

 * table_block_relation_estimate_size

 *

 * This function can't be directly used as the implementation of the

 * relation_estimate_size callback, because it has a few additional parameters.

 * Instead, it is intended to be used as a helper function; the caller can

 * pass through the arguments to its relation_estimate_size function plus the

 * additional values required here.

 *

 * overhead_bytes_per_tuple should contain the approximate number of bytes

 * of storage required to store a tuple above and beyond what is required for

 * the tuple data proper. Typically, this would include things like the

 * size of the tuple header and item pointer. This is only used for query

 * planning, so a table AM where the value is not constant could choose to

 * pass a "best guess".

 *

 * usable_bytes_per_page should contain the approximate number of bytes per

 * page usable for tuple data, excluding the page header and any anticipated

 * special space.

 */

void

table_block_relation_estimate_size(Relation rel, int32 *attr_widths,

                   BlockNumber *pages, double *tuples,

                   double *allvisfrac,

                   Size overhead_bytes_per_tuple,

                   Size usable_bytes_per_page)

{

  BlockNumber curpages;

  BlockNumber relpages;

  double    reltuples;

  BlockNumber relallvisible;

  double    density;

  /* it should have storage, so we can call the smgr */

  curpages = RelationGetNumberOfBlocks(rel);

  /* coerce values in pg_class to more desirable types */

  relpages = (BlockNumber) rel->rd_rel->relpages;

  reltuples = (double) rel->rd_rel->reltuples;

  relallvisible = (BlockNumber) rel->rd_rel->relallvisible;

  /*

   * HACK: if the relation has never yet been vacuumed, use a minimum size

   * estimate of 10 pages.  The idea here is to avoid assuming a

   * newly-created table is really small, even if it currently is, because

   * that may not be true once some data gets loaded into it.  Once a vacuum

   * or analyze cycle has been done on it, it's more reasonable to believe

   * the size is somewhat stable.

   *

   * (Note that this is only an issue if the plan gets cached and used again

   * after the table has been filled.  What we're trying to avoid is using a

   * nestloop-type plan on a table that has grown substantially since the

   * plan was made.  Normally, autovacuum/autoanalyze will occur once enough

   * inserts have happened and cause cached-plan invalidation; but that

   * doesn't happen instantaneously, and it won't happen at all for cases

   * such as temporary tables.)

   *

   * We test "never vacuumed" by seeing whether reltuples < 0.

   *

   * If the table has inheritance children, we don't apply this heuristic.

   * Totally empty parent tables are quite common, so we should be willing

   * to believe that they are empty.

   */

  if (curpages < 10 &&

    reltuples < 0 &&

    !rel->rd_rel->relhassubclass)

    curpages = 10;

  /* report estimated # pages */

  *pages = curpages;

  /* quick exit if rel is clearly empty */

  if (curpages == 0)

  {

    *tuples = 0;

    *allvisfrac = 0;

    return;

  }

  /* estimate number of tuples from previous tuple density */

  if (reltuples >= 0 && relpages > 0)

    density = reltuples / (double) relpages;

  else

  {

    /*

     * When we have no data because the relation was never yet vacuumed,

     * estimate tuple width from attribute datatypes.  We assume here that

     * the pages are completely full, which is OK for tables but is

     * probably an overestimate for indexes.  Fortunately

     * get_relation_info() can clamp the overestimate to the parent

     * table's size.

     *

     * Note: this code intentionally disregards alignment considerations,

     * because (a) that would be gilding the lily considering how crude

     * the estimate is, (b) it creates platform dependencies in the

     * default plans which are kind of a headache for regression testing,

     * and (c) different table AMs might use different padding schemes.

     */

    int32   tuple_width;

    int     fillfactor;

    /*

     * Without reltuples/relpages, we also need to consider fillfactor.

     * The other branch considers it implicitly by calculating density

     * from actual relpages/reltuples statistics.

     */

    fillfactor = RelationGetFillFactor(rel, HEAP_DEFAULT_FILLFACTOR);

    tuple_width = get_rel_data_width(rel, attr_widths);

    tuple_width += overhead_bytes_per_tuple;

    /* note: integer division is intentional here */

    density = (usable_bytes_per_page * fillfactor / 100) / tuple_width;

    /* There's at least one row on the page, even with low fillfactor. */

    density = clamp_row_est(density);

  }

  *tuples = rint(density * (double) curpages);

  /*

   * We use relallvisible as-is, rather than scaling it up like we do for

   * the pages and tuples counts, on the theory that any pages added since

   * the last VACUUM are most likely not marked all-visible.  But costsize.c

   * wants it converted to a fraction.

   */

  if (relallvisible == 0 || curpages <= 0)

    *allvisfrac = 0;

  else if ((double) relallvisible >= curpages)

    *allvisfrac = 1;

  else

    *allvisfrac = (double) relallvisible / curpages;

}