granicus.if.org Git - postgresql/blob - src/backend/commands/analyze.c

   1 /*-------------------------------------------------------------------------
   2  *
   3  * analyze.c
   4  *        the postgres statistics generator
   5  *
   6  * Portions Copyright (c) 1996-2001, PostgreSQL Global Development Group
   7  * Portions Copyright (c) 1994, Regents of the University of California
   8  *
   9  *
  10  * IDENTIFICATION
  11  *        $Header: /cvsroot/pgsql/src/backend/commands/analyze.c,v 1.27 2002/03/02 21:39:22 momjian Exp $
  12  *
  13  *-------------------------------------------------------------------------
  14  */
  15 #include "postgres.h"
  16
  17 #include <math.h>
  18
  19 #include "access/heapam.h"
  20 #include "access/tuptoaster.h"
  21 #include "catalog/catname.h"
  22 #include "catalog/indexing.h"
  23 #include "catalog/pg_operator.h"
  24 #include "catalog/pg_statistic.h"
  25 #include "catalog/pg_type.h"
  26 #include "commands/vacuum.h"
  27 #include "miscadmin.h"
  28 #include "parser/parse_oper.h"
  29 #include "utils/acl.h"
  30 #include "utils/builtins.h"
  31 #include "utils/datum.h"
  32 #include "utils/fmgroids.h"
  33 #include "utils/syscache.h"
  34 #include "utils/tuplesort.h"
  35
  36
  37 /*
  38  * Analysis algorithms supported
  39  */
  40 typedef enum
  41 {
  42         ALG_MINIMAL = 1,                        /* Compute only most-common-values */
  43         ALG_SCALAR                                      /* Compute MCV, histogram, sort
  44                                                                  * correlation */
  45 } AlgCode;
  46
  47 /*
  48  * To avoid consuming too much memory during analysis and/or too much space
  49  * in the resulting pg_statistic rows, we ignore varlena datums that are wider
  50  * than WIDTH_THRESHOLD (after detoasting!).  This is legitimate for MCV
  51  * and distinct-value calculations since a wide value is unlikely to be
  52  * duplicated at all, much less be a most-common value.  For the same reason,
  53  * ignoring wide values will not affect our estimates of histogram bin
  54  * boundaries very much.
  55  */
  56 #define WIDTH_THRESHOLD  256
  57
  58 /*
  59  * We build one of these structs for each attribute (column) that is to be
  60  * analyzed.  The struct and subsidiary data are in TransactionCommandContext,
  61  * so they live until the end of the ANALYZE operation.
  62  */
  63 typedef struct
  64 {
  65         /* These fields are set up by examine_attribute */
  66         int                     attnum;                 /* attribute number */
  67         AlgCode         algcode;                /* Which algorithm to use for this column */
  68         int                     minrows;                /* Minimum # of rows wanted for stats */
  69         Form_pg_attribute attr;         /* copy of pg_attribute row for column */
  70         Form_pg_type attrtype;          /* copy of pg_type row for column */
  71         Oid                     eqopr;                  /* '=' operator for datatype, if any */
  72         Oid                     eqfunc;                 /* and associated function */
  73         Oid                     ltopr;                  /* '<' operator for datatype, if any */
  74
  75         /*
  76          * These fields are filled in by the actual statistics-gathering
  77          * routine
  78          */
  79         bool            stats_valid;
  80         float4          stanullfrac;    /* fraction of entries that are NULL */
  81         int4            stawidth;               /* average width */
  82         float4          stadistinct;    /* # distinct values */
  83         int2            stakind[STATISTIC_NUM_SLOTS];
  84         Oid                     staop[STATISTIC_NUM_SLOTS];
  85         int                     numnumbers[STATISTIC_NUM_SLOTS];
  86         float4     *stanumbers[STATISTIC_NUM_SLOTS];
  87         int                     numvalues[STATISTIC_NUM_SLOTS];
  88         Datum      *stavalues[STATISTIC_NUM_SLOTS];
  89 } VacAttrStats;
  90
  91
  92 typedef struct
  93 {
  94         Datum           value;                  /* a data value */
  95         int                     tupno;                  /* position index for tuple it came from */
  96 } ScalarItem;
  97
  98 typedef struct
  99 {
 100         int                     count;                  /* # of duplicates */
 101         int                     first;                  /* values[] index of first occurrence */
 102 } ScalarMCVItem;
 103
 104
 105 #define swapInt(a,b)    do {int _tmp; _tmp=a; a=b; b=_tmp;} while(0)
 106 #define swapDatum(a,b)  do {Datum _tmp; _tmp=a; a=b; b=_tmp;} while(0)
 107
 108 static int elevel = -1;
 109
 110 /* context information for compare_scalars() */
 111 static FmgrInfo *datumCmpFn;
 112 static SortFunctionKind datumCmpFnKind;
 113 static int *datumCmpTupnoLink;
 114
 115
 116 static VacAttrStats *examine_attribute(Relation onerel, int attnum);
 117 static int acquire_sample_rows(Relation onerel, HeapTuple *rows,
 118                                         int targrows, double *totalrows);
 119 static double random_fract(void);
 120 static double init_selection_state(int n);
 121 static double select_next_random_record(double t, int n, double *stateptr);
 122 static int      compare_rows(const void *a, const void *b);
 123 static int      compare_scalars(const void *a, const void *b);
 124 static int      compare_mcvs(const void *a, const void *b);
 125 static void compute_minimal_stats(VacAttrStats *stats,
 126                                           TupleDesc tupDesc, double totalrows,
 127                                           HeapTuple *rows, int numrows);
 128 static void compute_scalar_stats(VacAttrStats *stats,
 129                                          TupleDesc tupDesc, double totalrows,
 130                                          HeapTuple *rows, int numrows);
 131 static void update_attstats(Oid relid, int natts, VacAttrStats **vacattrstats);
 132
 133
 134 /*
 135  *      analyze_rel() -- analyze one relation
 136  */
 137 void
 138 analyze_rel(Oid relid, VacuumStmt *vacstmt)
 139 {
 140         Relation        onerel;
 141         Form_pg_attribute *attr;
 142         int                     attr_cnt,
 143                                 tcnt,
 144                                 i;
 145         VacAttrStats **vacattrstats;
 146         int                     targrows,
 147                                 numrows;
 148         double          totalrows;
 149         HeapTuple  *rows;
 150         HeapTuple       tuple;
 151
 152         if (vacstmt->verbose)
 153                 elevel = INFO;
 154         else
 155                 elevel = DEBUG1;
 156
 157         /*
 158          * Begin a transaction for analyzing this relation.
 159          *
 160          * Note: All memory allocated during ANALYZE will live in
 161          * TransactionCommandContext or a subcontext thereof, so it will all
 162          * be released by transaction commit at the end of this routine.
 163          */
 164         StartTransactionCommand();
 165
 166         /*
 167          * Check for user-requested abort.      Note we want this to be inside a
 168          * transaction, so xact.c doesn't issue useless NOTICE.
 169          */
 170         CHECK_FOR_INTERRUPTS();
 171
 172         /*
 173          * Race condition -- if the pg_class tuple has gone away since the
 174          * last time we saw it, we don't need to process it.
 175          */
 176         tuple = SearchSysCache(RELOID,
 177                                                    ObjectIdGetDatum(relid),
 178                                                    0, 0, 0);
 179         if (!HeapTupleIsValid(tuple))
 180         {
 181                 CommitTransactionCommand();
 182                 return;
 183         }
 184
 185         /*
 186          * We can ANALYZE any table except pg_statistic. See update_attstats
 187          */
 188         if (strcmp(NameStr(((Form_pg_class) GETSTRUCT(tuple))->relname),
 189                            StatisticRelationName) == 0)
 190         {
 191                 ReleaseSysCache(tuple);
 192                 CommitTransactionCommand();
 193                 return;
 194         }
 195         ReleaseSysCache(tuple);
 196
 197         /*
 198          * Open the class, getting only a read lock on it, and check
 199          * permissions. Permissions check should match vacuum's check!
 200          */
 201         onerel = heap_open(relid, AccessShareLock);
 202
 203         if (!(pg_ownercheck(GetUserId(), RelationGetRelationName(onerel),
 204                                                 RELNAME) ||
 205                   (is_dbadmin(MyDatabaseId) && !onerel->rd_rel->relisshared)))
 206         {
 207                 /* No need for a notice if we already complained during VACUUM */
 208                 if (!vacstmt->vacuum)
 209                         elog(NOTICE, "Skipping \"%s\" --- only table or database owner can ANALYZE it",
 210                                  RelationGetRelationName(onerel));
 211                 heap_close(onerel, NoLock);
 212                 CommitTransactionCommand();
 213                 return;
 214         }
 215
 216         elog(elevel, "Analyzing %s", RelationGetRelationName(onerel));
 217
 218         /*
 219          * Determine which columns to analyze
 220          *
 221          * Note that system attributes are never analyzed.
 222          */
 223         attr = onerel->rd_att->attrs;
 224         attr_cnt = onerel->rd_att->natts;
 225
 226         if (vacstmt->va_cols != NIL)
 227         {
 228                 List       *le;
 229
 230                 vacattrstats = (VacAttrStats **) palloc(length(vacstmt->va_cols) *
 231                                                                                                 sizeof(VacAttrStats *));
 232                 tcnt = 0;
 233                 foreach(le, vacstmt->va_cols)
 234                 {
 235                         char       *col = strVal(lfirst(le));
 236
 237                         for (i = 0; i < attr_cnt; i++)
 238                         {
 239                                 if (namestrcmp(&(attr[i]->attname), col) == 0)
 240                                         break;
 241                         }
 242                         if (i >= attr_cnt)
 243                                 elog(ERROR, "ANALYZE: there is no attribute %s in %s",
 244                                          col, RelationGetRelationName(onerel));
 245                         vacattrstats[tcnt] = examine_attribute(onerel, i + 1);
 246                         if (vacattrstats[tcnt] != NULL)
 247                                 tcnt++;
 248                 }
 249                 attr_cnt = tcnt;
 250         }
 251         else
 252         {
 253                 vacattrstats = (VacAttrStats **) palloc(attr_cnt *
 254                                                                                                 sizeof(VacAttrStats *));
 255                 tcnt = 0;
 256                 for (i = 0; i < attr_cnt; i++)
 257                 {
 258                         vacattrstats[tcnt] = examine_attribute(onerel, i + 1);
 259                         if (vacattrstats[tcnt] != NULL)
 260                                 tcnt++;
 261                 }
 262                 attr_cnt = tcnt;
 263         }
 264
 265         /*
 266          * Quit if no analyzable columns
 267          */
 268         if (attr_cnt <= 0)
 269         {
 270                 heap_close(onerel, NoLock);
 271                 CommitTransactionCommand();
 272                 return;
 273         }
 274
 275         /*
 276          * Determine how many rows we need to sample, using the worst case
 277          * from all analyzable columns.  We use a lower bound of 100 rows to
 278          * avoid possible overflow in Vitter's algorithm.
 279          */
 280         targrows = 100;
 281         for (i = 0; i < attr_cnt; i++)
 282         {
 283                 if (targrows < vacattrstats[i]->minrows)
 284                         targrows = vacattrstats[i]->minrows;
 285         }
 286
 287         /*
 288          * Acquire the sample rows
 289          */
 290         rows = (HeapTuple *) palloc(targrows * sizeof(HeapTuple));
 291         numrows = acquire_sample_rows(onerel, rows, targrows, &totalrows);
 292
 293         /*
 294          * If we are running a standalone ANALYZE, update pages/tuples stats
 295          * in pg_class.  We have the accurate page count from heap_beginscan,
 296          * but only an approximate number of tuples; therefore, if we are part
 297          * of VACUUM ANALYZE do *not* overwrite the accurate count already
 298          * inserted by VACUUM.
 299          */
 300         if (!vacstmt->vacuum)
 301                 vac_update_relstats(RelationGetRelid(onerel),
 302                                                         onerel->rd_nblocks,
 303                                                         totalrows,
 304                                                         RelationGetForm(onerel)->relhasindex);
 305
 306         /*
 307          * Compute the statistics.      Temporary results during the calculations
 308          * for each column are stored in a child context.  The calc routines
 309          * are responsible to make sure that whatever they store into the
 310          * VacAttrStats structure is allocated in TransactionCommandContext.
 311          */
 312         if (numrows > 0)
 313         {
 314                 MemoryContext col_context,
 315                                         old_context;
 316
 317                 col_context = AllocSetContextCreate(CurrentMemoryContext,
 318                                                                                         "Analyze Column",
 319                                                                                         ALLOCSET_DEFAULT_MINSIZE,
 320                                                                                         ALLOCSET_DEFAULT_INITSIZE,
 321                                                                                         ALLOCSET_DEFAULT_MAXSIZE);
 322                 old_context = MemoryContextSwitchTo(col_context);
 323                 for (i = 0; i < attr_cnt; i++)
 324                 {
 325                         switch (vacattrstats[i]->algcode)
 326                         {
 327                                 case ALG_MINIMAL:
 328                                         compute_minimal_stats(vacattrstats[i],
 329                                                                                   onerel->rd_att, totalrows,
 330                                                                                   rows, numrows);
 331                                         break;
 332                                 case ALG_SCALAR:
 333                                         compute_scalar_stats(vacattrstats[i],
 334                                                                                  onerel->rd_att, totalrows,
 335                                                                                  rows, numrows);
 336                                         break;
 337                         }
 338                         MemoryContextResetAndDeleteChildren(col_context);
 339                 }
 340                 MemoryContextSwitchTo(old_context);
 341                 MemoryContextDelete(col_context);
 342
 343                 /*
 344                  * Emit the completed stats rows into pg_statistic, replacing any
 345                  * previous statistics for the target columns.  (If there are
 346                  * stats in pg_statistic for columns we didn't process, we leave
 347                  * them alone.)
 348                  */
 349                 update_attstats(relid, attr_cnt, vacattrstats);
 350         }
 351
 352         /*
 353          * Close source relation now, but keep lock so that no one deletes it
 354          * before we commit.  (If someone did, they'd fail to clean up the
 355          * entries we made in pg_statistic.)
 356          */
 357         heap_close(onerel, NoLock);
 358
 359         /* Commit and release working memory */
 360         CommitTransactionCommand();
 361 }
 362
 363 /*
 364  * examine_attribute -- pre-analysis of a single column
 365  *
 366  * Determine whether the column is analyzable; if so, create and initialize
 367  * a VacAttrStats struct for it.  If not, return NULL.
 368  */
 369 static VacAttrStats *
 370 examine_attribute(Relation onerel, int attnum)
 371 {
 372         Form_pg_attribute attr = onerel->rd_att->attrs[attnum - 1];
 373         Operator        func_operator;
 374         Oid                     oprrest;
 375         HeapTuple       typtuple;
 376         Oid                     eqopr = InvalidOid;
 377         Oid                     eqfunc = InvalidOid;
 378         Oid                     ltopr = InvalidOid;
 379         VacAttrStats *stats;
 380
 381         /* Don't analyze column if user has specified not to */
 382         if (attr->attstattarget <= 0)
 383                 return NULL;
 384
 385         /* If column has no "=" operator, we can't do much of anything */
 386         func_operator = compatible_oper("=",
 387                                                                         attr->atttypid,
 388                                                                         attr->atttypid,
 389                                                                         true);
 390         if (func_operator != NULL)
 391         {
 392                 oprrest = ((Form_pg_operator) GETSTRUCT(func_operator))->oprrest;
 393                 if (oprrest == F_EQSEL)
 394                 {
 395                         eqopr = oprid(func_operator);
 396                         eqfunc = oprfuncid(func_operator);
 397                 }
 398                 ReleaseSysCache(func_operator);
 399         }
 400         if (!OidIsValid(eqfunc))
 401                 return NULL;
 402
 403         /*
 404          * If we have "=" then we're at least able to do the minimal
 405          * algorithm, so start filling in a VacAttrStats struct.
 406          */
 407         stats = (VacAttrStats *) palloc(sizeof(VacAttrStats));
 408         MemSet(stats, 0, sizeof(VacAttrStats));
 409         stats->attnum = attnum;
 410         stats->attr = (Form_pg_attribute) palloc(ATTRIBUTE_TUPLE_SIZE);
 411         memcpy(stats->attr, attr, ATTRIBUTE_TUPLE_SIZE);
 412         typtuple = SearchSysCache(TYPEOID,
 413                                                           ObjectIdGetDatum(attr->atttypid),
 414                                                           0, 0, 0);
 415         if (!HeapTupleIsValid(typtuple))
 416                 elog(ERROR, "cache lookup of type %u failed", attr->atttypid);
 417         stats->attrtype = (Form_pg_type) palloc(sizeof(FormData_pg_type));
 418         memcpy(stats->attrtype, GETSTRUCT(typtuple), sizeof(FormData_pg_type));
 419         ReleaseSysCache(typtuple);
 420         stats->eqopr = eqopr;
 421         stats->eqfunc = eqfunc;
 422
 423         /* Is there a "<" operator with suitable semantics? */
 424         func_operator = compatible_oper("<",
 425                                                                         attr->atttypid,
 426                                                                         attr->atttypid,
 427                                                                         true);
 428         if (func_operator != NULL)
 429         {
 430                 oprrest = ((Form_pg_operator) GETSTRUCT(func_operator))->oprrest;
 431                 if (oprrest == F_SCALARLTSEL)
 432                         ltopr = oprid(func_operator);
 433                 ReleaseSysCache(func_operator);
 434         }
 435         stats->ltopr = ltopr;
 436
 437         /*
 438          * Determine the algorithm to use (this will get more complicated
 439          * later)
 440          */
 441         if (OidIsValid(ltopr))
 442         {
 443                 /* Seems to be a scalar datatype */
 444                 stats->algcode = ALG_SCALAR;
 445                 /*--------------------
 446                  * The following choice of minrows is based on the paper
 447                  * "Random sampling for histogram construction: how much is enough?"
 448                  * by Surajit Chaudhuri, Rajeev Motwani and Vivek Narasayya, in
 449                  * Proceedings of ACM SIGMOD International Conference on Management
 450                  * of Data, 1998, Pages 436-447.  Their Corollary 1 to Theorem 5
 451                  * says that for table size n, histogram size k, maximum relative
 452                  * error in bin size f, and error probability gamma, the minimum
 453                  * random sample size is
 454                  *              r = 4 * k * ln(2*n/gamma) / f^2
 455                  * Taking f = 0.5, gamma = 0.01, n = 1 million rows, we obtain
 456                  *              r = 305.82 * k
 457                  * Note that because of the log function, the dependence on n is
 458                  * quite weak; even at n = 1 billion, a 300*k sample gives <= 0.59
 459                  * bin size error with probability 0.99.  So there's no real need to
 460                  * scale for n, which is a good thing because we don't necessarily
 461                  * know it at this point.
 462                  *--------------------
 463                  */
 464                 stats->minrows = 300 * attr->attstattarget;
 465         }
 466         else
 467         {
 468                 /* Can't do much but the minimal stuff */
 469                 stats->algcode = ALG_MINIMAL;
 470                 /* Might as well use the same minrows as above */
 471                 stats->minrows = 300 * attr->attstattarget;
 472         }
 473
 474         return stats;
 475 }
 476
 477 /*
 478  * acquire_sample_rows -- acquire a random sample of rows from the table
 479  *
 480  * Up to targrows rows are collected (if there are fewer than that many
 481  * rows in the table, all rows are collected).  When the table is larger
 482  * than targrows, a truly random sample is collected: every row has an
 483  * equal chance of ending up in the final sample.
 484  *
 485  * We also estimate the total number of rows in the table, and return that
 486  * into *totalrows.
 487  *
 488  * The returned list of tuples is in order by physical position in the table.
 489  * (We will rely on this later to derive correlation estimates.)
 490  */
 491 static int
 492 acquire_sample_rows(Relation onerel, HeapTuple *rows, int targrows,
 493                                         double *totalrows)
 494 {
 495         int                     numrows = 0;
 496         HeapScanDesc scan;
 497         HeapTuple       tuple;
 498         ItemPointer lasttuple;
 499         BlockNumber lastblock,
 500                                 estblock;
 501         OffsetNumber lastoffset;
 502         int                     numest;
 503         double          tuplesperpage;
 504         double          t;
 505         double          rstate;
 506
 507         Assert(targrows > 1);
 508
 509         /*
 510          * Do a simple linear scan until we reach the target number of rows.
 511          */
 512         scan = heap_beginscan(onerel, false, SnapshotNow, 0, NULL);
 513         while (HeapTupleIsValid(tuple = heap_getnext(scan, 0)))
 514         {
 515                 rows[numrows++] = heap_copytuple(tuple);
 516                 if (numrows >= targrows)
 517                         break;
 518                 CHECK_FOR_INTERRUPTS();
 519         }
 520         heap_endscan(scan);
 521
 522         /*
 523          * If we ran out of tuples then we're done, no matter how few we
 524          * collected.  No sort is needed, since they're already in order.
 525          */
 526         if (!HeapTupleIsValid(tuple))
 527         {
 528                 *totalrows = (double) numrows;
 529                 return numrows;
 530         }
 531
 532         /*
 533          * Otherwise, start replacing tuples in the sample until we reach the
 534          * end of the relation.  This algorithm is from Jeff Vitter's paper
 535          * (see full citation below).  It works by repeatedly computing the
 536          * number of the next tuple we want to fetch, which will replace a
 537          * randomly chosen element of the reservoir (current set of tuples).
 538          * At all times the reservoir is a true random sample of the tuples
 539          * we've passed over so far, so when we fall off the end of the
 540          * relation we're done.
 541          *
 542          * A slight difficulty is that since we don't want to fetch tuples or
 543          * even pages that we skip over, it's not possible to fetch *exactly*
 544          * the N'th tuple at each step --- we don't know how many valid tuples
 545          * are on the skipped pages.  We handle this by assuming that the
 546          * average number of valid tuples/page on the pages already scanned
 547          * over holds good for the rest of the relation as well; this lets us
 548          * estimate which page the next tuple should be on and its position in
 549          * the page.  Then we fetch the first valid tuple at or after that
 550          * position, being careful not to use the same tuple twice.  This
 551          * approach should still give a good random sample, although it's not
 552          * perfect.
 553          */
 554         lasttuple = &(rows[numrows - 1]->t_self);
 555         lastblock = ItemPointerGetBlockNumber(lasttuple);
 556         lastoffset = ItemPointerGetOffsetNumber(lasttuple);
 557
 558         /*
 559          * If possible, estimate tuples/page using only completely-scanned
 560          * pages.
 561          */
 562         for (numest = numrows; numest > 0; numest--)
 563         {
 564                 if (ItemPointerGetBlockNumber(&(rows[numest - 1]->t_self)) != lastblock)
 565                         break;
 566         }
 567         if (numest == 0)
 568         {
 569                 numest = numrows;               /* don't have a full page? */
 570                 estblock = lastblock + 1;
 571         }
 572         else
 573                 estblock = lastblock;
 574         tuplesperpage = (double) numest / (double) estblock;
 575
 576         t = (double) numrows;           /* t is the # of records processed so far */
 577         rstate = init_selection_state(targrows);
 578         for (;;)
 579         {
 580                 double          targpos;
 581                 BlockNumber targblock;
 582                 Buffer          targbuffer;
 583                 Page            targpage;
 584                 OffsetNumber targoffset,
 585                                         maxoffset;
 586
 587                 CHECK_FOR_INTERRUPTS();
 588
 589                 t = select_next_random_record(t, targrows, &rstate);
 590                 /* Try to read the t'th record in the table */
 591                 targpos = t / tuplesperpage;
 592                 targblock = (BlockNumber) targpos;
 593                 targoffset = ((int) ((targpos - targblock) * tuplesperpage)) +
 594                         FirstOffsetNumber;
 595                 /* Make sure we are past the last selected record */
 596                 if (targblock <= lastblock)
 597                 {
 598                         targblock = lastblock;
 599                         if (targoffset <= lastoffset)
 600                                 targoffset = lastoffset + 1;
 601                 }
 602                 /* Loop to find first valid record at or after given position */
 603 pageloop:;
 604
 605                 /*
 606                  * Have we fallen off the end of the relation?  (We rely on
 607                  * heap_beginscan to have updated rd_nblocks.)
 608                  */
 609                 if (targblock >= onerel->rd_nblocks)
 610                         break;
 611
 612                 /*
 613                  * We must maintain a pin on the target page's buffer to ensure
 614                  * that the maxoffset value stays good (else concurrent VACUUM
 615                  * might delete tuples out from under us).      Hence, pin the page
 616                  * until we are done looking at it.  We don't maintain a lock on
 617                  * the page, so tuples could get added to it, but we ignore such
 618                  * tuples.
 619                  */
 620                 targbuffer = ReadBuffer(onerel, targblock);
 621                 if (!BufferIsValid(targbuffer))
 622                         elog(ERROR, "acquire_sample_rows: ReadBuffer(%s,%u) failed",
 623                                  RelationGetRelationName(onerel), targblock);
 624                 LockBuffer(targbuffer, BUFFER_LOCK_SHARE);
 625                 targpage = BufferGetPage(targbuffer);
 626                 maxoffset = PageGetMaxOffsetNumber(targpage);
 627                 LockBuffer(targbuffer, BUFFER_LOCK_UNLOCK);
 628
 629                 for (;;)
 630                 {
 631                         HeapTupleData targtuple;
 632                         Buffer          tupbuffer;
 633
 634                         if (targoffset > maxoffset)
 635                         {
 636                                 /* Fell off end of this page, try next */
 637                                 ReleaseBuffer(targbuffer);
 638                                 targblock++;
 639                                 targoffset = FirstOffsetNumber;
 640                                 goto pageloop;
 641                         }
 642                         ItemPointerSet(&targtuple.t_self, targblock, targoffset);
 643                         heap_fetch(onerel, SnapshotNow, &targtuple, &tupbuffer, NULL);
 644                         if (targtuple.t_data != NULL)
 645                         {
 646                                 /*
 647                                  * Found a suitable tuple, so save it, replacing one old
 648                                  * tuple at random
 649                                  */
 650                                 int                     k = (int) (targrows * random_fract());
 651
 652                                 Assert(k >= 0 && k < targrows);
 653                                 heap_freetuple(rows[k]);
 654                                 rows[k] = heap_copytuple(&targtuple);
 655                                 /* this releases the second pin acquired by heap_fetch: */
 656                                 ReleaseBuffer(tupbuffer);
 657                                 /* this releases the initial pin: */
 658                                 ReleaseBuffer(targbuffer);
 659                                 lastblock = targblock;
 660                                 lastoffset = targoffset;
 661                                 break;
 662                         }
 663                         /* this tuple is dead, so advance to next one on same page */
 664                         targoffset++;
 665                 }
 666         }
 667
 668         /*
 669          * Now we need to sort the collected tuples by position (itempointer).
 670          */
 671         qsort((void *) rows, numrows, sizeof(HeapTuple), compare_rows);
 672
 673         /*
 674          * Estimate total number of valid rows in relation.
 675          */
 676         *totalrows = floor((double) onerel->rd_nblocks * tuplesperpage + 0.5);
 677
 678         return numrows;
 679 }
 680
 681 /* Select a random value R uniformly distributed in 0 < R < 1 */
 682 static double
 683 random_fract(void)
 684 {
 685         long            z;
 686
 687         /* random() can produce endpoint values, try again if so */
 688         do
 689         {
 690                 z = random();
 691         } while (!(z > 0 && z < MAX_RANDOM_VALUE));
 692         return (double) z / (double) MAX_RANDOM_VALUE;
 693 }
 694
 695 /*
 696  * These two routines embody Algorithm Z from "Random sampling with a
 697  * reservoir" by Jeffrey S. Vitter, in ACM Trans. Math. Softw. 11, 1
 698  * (Mar. 1985), Pages 37-57.  While Vitter describes his algorithm in terms
 699  * of the count S of records to skip before processing another record,
 700  * it is convenient to work primarily with t, the index (counting from 1)
 701  * of the last record processed and next record to process.  The only extra
 702  * state needed between calls is W, a random state variable.
 703  *
 704  * Note: the original algorithm defines t, S, numer, and denom as integers.
 705  * Here we express them as doubles to avoid overflow if the number of rows
 706  * in the table exceeds INT_MAX.  The algorithm should work as long as the
 707  * row count does not become so large that it is not represented accurately
 708  * in a double (on IEEE-math machines this would be around 2^52 rows).
 709  *
 710  * init_selection_state computes the initial W value.
 711  *
 712  * Given that we've already processed t records (t >= n),
 713  * select_next_random_record determines the number of the next record to
 714  * process.
 715  */
 716 static double
 717 init_selection_state(int n)
 718 {
 719         /* Initial value of W (for use when Algorithm Z is first applied) */
 720         return exp(-log(random_fract()) / n);
 721 }
 722
 723 static double
 724 select_next_random_record(double t, int n, double *stateptr)
 725 {
 726         /* The magic constant here is T from Vitter's paper */
 727         if (t <= (22.0 * n))
 728         {
 729                 /* Process records using Algorithm X until t is large enough */
 730                 double          V,
 731                                         quot;
 732
 733                 V = random_fract();             /* Generate V */
 734                 t += 1;
 735                 quot = (t - (double) n) / t;
 736                 /* Find min S satisfying (4.1) */
 737                 while (quot > V)
 738                 {
 739                         t += 1;
 740                         quot *= (t - (double) n) / t;
 741                 }
 742         }
 743         else
 744         {
 745                 /* Now apply Algorithm Z */
 746                 double          W = *stateptr;
 747                 double          term = t - (double) n + 1;
 748                 double          S;
 749
 750                 for (;;)
 751                 {
 752                         double          numer,
 753                                                 numer_lim,
 754                                                 denom;
 755                         double          U,
 756                                                 X,
 757                                                 lhs,
 758                                                 rhs,
 759                                                 y,
 760                                                 tmp;
 761
 762                         /* Generate U and X */
 763                         U = random_fract();
 764                         X = t * (W - 1.0);
 765                         S = floor(X);           /* S is tentatively set to floor(X) */
 766                         /* Test if U <= h(S)/cg(X) in the manner of (6.3) */
 767                         tmp = (t + 1) / term;
 768                         lhs = exp(log(((U * tmp * tmp) * (term + S)) / (t + X)) / n);
 769                         rhs = (((t + X) / (term + S)) * term) / t;
 770                         if (lhs <= rhs)
 771                         {
 772                                 W = rhs / lhs;
 773                                 break;
 774                         }
 775                         /* Test if U <= f(S)/cg(X) */
 776                         y = (((U * (t + 1)) / term) * (t + S + 1)) / (t + X);
 777                         if ((double) n < S)
 778                         {
 779                                 denom = t;
 780                                 numer_lim = term + S;
 781                         }
 782                         else
 783                         {
 784                                 denom = t - (double) n + S;
 785                                 numer_lim = t + 1;
 786                         }
 787                         for (numer = t + S; numer >= numer_lim; numer -= 1)
 788                         {
 789                                 y *= numer / denom;
 790                                 denom -= 1;
 791                         }
 792                         W = exp(-log(random_fract()) / n);      /* Generate W in advance */
 793                         if (exp(log(y) / n) <= (t + X) / t)
 794                                 break;
 795                 }
 796                 t += S + 1;
 797                 *stateptr = W;
 798         }
 799         return t;
 800 }
 801
 802 /*
 803  * qsort comparator for sorting rows[] array
 804  */
 805 static int
 806 compare_rows(const void *a, const void *b)
 807 {
 808         HeapTuple       ha = *(HeapTuple *) a;
 809         HeapTuple       hb = *(HeapTuple *) b;
 810         BlockNumber ba = ItemPointerGetBlockNumber(&ha->t_self);
 811         OffsetNumber oa = ItemPointerGetOffsetNumber(&ha->t_self);
 812         BlockNumber bb = ItemPointerGetBlockNumber(&hb->t_self);
 813         OffsetNumber ob = ItemPointerGetOffsetNumber(&hb->t_self);
 814
 815         if (ba < bb)
 816                 return -1;
 817         if (ba > bb)
 818                 return 1;
 819         if (oa < ob)
 820                 return -1;
 821         if (oa > ob)
 822                 return 1;
 823         return 0;
 824 }
 825
 826
 827 /*
 828  *      compute_minimal_stats() -- compute minimal column statistics
 829  *
 830  *      We use this when we can find only an "=" operator for the datatype.
 831  *
 832  *      We determine the fraction of non-null rows, the average width, the
 833  *      most common values, and the (estimated) number of distinct values.
 834  *
 835  *      The most common values are determined by brute force: we keep a list
 836  *      of previously seen values, ordered by number of times seen, as we scan
 837  *      the samples.  A newly seen value is inserted just after the last
 838  *      multiply-seen value, causing the bottommost (oldest) singly-seen value
 839  *      to drop off the list.  The accuracy of this method, and also its cost,
 840  *      depend mainly on the length of the list we are willing to keep.
 841  */
 842 static void
 843 compute_minimal_stats(VacAttrStats *stats,
 844                                           TupleDesc tupDesc, double totalrows,
 845                                           HeapTuple *rows, int numrows)
 846 {
 847         int                     i;
 848         int                     null_cnt = 0;
 849         int                     nonnull_cnt = 0;
 850         int                     toowide_cnt = 0;
 851         double          total_width = 0;
 852         bool            is_varlena = (!stats->attr->attbyval &&
 853                                                           stats->attr->attlen == -1);
 854         FmgrInfo        f_cmpeq;
 855         typedef struct
 856         {
 857                 Datum           value;
 858                 int                     count;
 859         } TrackItem;
 860         TrackItem  *track;
 861         int                     track_cnt,
 862                                 track_max;
 863         int                     num_mcv = stats->attr->attstattarget;
 864
 865         /*
 866          * We track up to 2*n values for an n-element MCV list; but at least
 867          * 10
 868          */
 869         track_max = 2 * num_mcv;
 870         if (track_max < 10)
 871                 track_max = 10;
 872         track = (TrackItem *) palloc(track_max * sizeof(TrackItem));
 873         track_cnt = 0;
 874
 875         fmgr_info(stats->eqfunc, &f_cmpeq);
 876
 877         for (i = 0; i < numrows; i++)
 878         {
 879                 HeapTuple       tuple = rows[i];
 880                 Datum           value;
 881                 bool            isnull;
 882                 bool            match;
 883                 int                     firstcount1,
 884                                         j;
 885
 886                 CHECK_FOR_INTERRUPTS();
 887
 888                 value = heap_getattr(tuple, stats->attnum, tupDesc, &isnull);
 889
 890                 /* Check for null/nonnull */
 891                 if (isnull)
 892                 {
 893                         null_cnt++;
 894                         continue;
 895                 }
 896                 nonnull_cnt++;
 897
 898                 /*
 899                  * If it's a varlena field, add up widths for average width
 900                  * calculation.  Note that if the value is toasted, we use the
 901                  * toasted width.  We don't bother with this calculation if it's a
 902                  * fixed-width type.
 903                  */
 904                 if (is_varlena)
 905                 {
 906                         total_width += VARSIZE(DatumGetPointer(value));
 907
 908                         /*
 909                          * If the value is toasted, we want to detoast it just once to
 910                          * avoid repeated detoastings and resultant excess memory
 911                          * usage during the comparisons.  Also, check to see if the
 912                          * value is excessively wide, and if so don't detoast at all
 913                          * --- just ignore the value.
 914                          */
 915                         if (toast_raw_datum_size(value) > WIDTH_THRESHOLD)
 916                         {
 917                                 toowide_cnt++;
 918                                 continue;
 919                         }
 920                         value = PointerGetDatum(PG_DETOAST_DATUM(value));
 921                 }
 922
 923                 /*
 924                  * See if the value matches anything we're already tracking.
 925                  */
 926                 match = false;
 927                 firstcount1 = track_cnt;
 928                 for (j = 0; j < track_cnt; j++)
 929                 {
 930                         if (DatumGetBool(FunctionCall2(&f_cmpeq, value, track[j].value)))
 931                         {
 932                                 match = true;
 933                                 break;
 934                         }
 935                         if (j < firstcount1 && track[j].count == 1)
 936                                 firstcount1 = j;
 937                 }
 938
 939                 if (match)
 940                 {
 941                         /* Found a match */
 942                         track[j].count++;
 943                         /* This value may now need to "bubble up" in the track list */
 944                         while (j > 0 && track[j].count > track[j - 1].count)
 945                         {
 946                                 swapDatum(track[j].value, track[j - 1].value);
 947                                 swapInt(track[j].count, track[j - 1].count);
 948                                 j--;
 949                         }
 950                 }
 951                 else
 952                 {
 953                         /* No match.  Insert at head of count-1 list */
 954                         if (track_cnt < track_max)
 955                                 track_cnt++;
 956                         for (j = track_cnt - 1; j > firstcount1; j--)
 957                         {
 958                                 track[j].value = track[j - 1].value;
 959                                 track[j].count = track[j - 1].count;
 960                         }
 961                         if (firstcount1 < track_cnt)
 962                         {
 963                                 track[firstcount1].value = value;
 964                                 track[firstcount1].count = 1;
 965                         }
 966                 }
 967         }
 968
 969         /* We can only compute valid stats if we found some non-null values. */
 970         if (nonnull_cnt > 0)
 971         {
 972                 int                     nmultiple,
 973                                         summultiple;
 974
 975                 stats->stats_valid = true;
 976                 /* Do the simple null-frac and width stats */
 977                 stats->stanullfrac = (double) null_cnt / (double) numrows;
 978                 if (is_varlena)
 979                         stats->stawidth = total_width / (double) nonnull_cnt;
 980                 else
 981                         stats->stawidth = stats->attrtype->typlen;
 982
 983                 /* Count the number of values we found multiple times */
 984                 summultiple = 0;
 985                 for (nmultiple = 0; nmultiple < track_cnt; nmultiple++)
 986                 {
 987                         if (track[nmultiple].count == 1)
 988                                 break;
 989                         summultiple += track[nmultiple].count;
 990                 }
 991
 992                 if (nmultiple == 0)
 993                 {
 994                         /* If we found no repeated values, assume it's a unique column */
 995                         stats->stadistinct = -1.0;
 996                 }
 997                 else if (track_cnt < track_max && toowide_cnt == 0 &&
 998                                  nmultiple == track_cnt)
 999                 {
1000                         /*
1001                          * Our track list includes every value in the sample, and
1002                          * every value appeared more than once.  Assume the column has
1003                          * just these values.
1004                          */
1005                         stats->stadistinct = track_cnt;
1006                 }
1007                 else
1008                 {
1009                         /*----------
1010                          * Estimate the number of distinct values using the estimator
1011                          * proposed by Haas and Stokes in IBM Research Report RJ 10025:
1012                          *              n*d / (n - f1 + f1*n/N)
1013                          * where f1 is the number of distinct values that occurred
1014                          * exactly once in our sample of n rows (from a total of N),
1015                          * and d is the total number of distinct values in the sample.
1016                          * This is their Duj1 estimator; the other estimators they
1017                          * recommend are considerably more complex, and are numerically
1018                          * very unstable when n is much smaller than N.
1019                          *
1020                          * We assume (not very reliably!) that all the multiply-occurring
1021                          * values are reflected in the final track[] list, and the other
1022                          * nonnull values all appeared but once.  (XXX this usually
1023                          * results in a drastic overestimate of ndistinct.      Can we do
1024                          * any better?)
1025                          *----------
1026                          */
1027                         int                     f1 = nonnull_cnt - summultiple;
1028                         int                     d = f1 + nmultiple;
1029                         double          numer, denom, stadistinct;
1030
1031                         numer = (double) numrows * (double) d;
1032                         denom = (double) (numrows - f1) +
1033                                 (double) f1 * (double) numrows / totalrows;
1034                         stadistinct = numer / denom;
1035                         /* Clamp to sane range in case of roundoff error */
1036                         if (stadistinct < (double) d)
1037                                 stadistinct = (double) d;
1038                         if (stadistinct > totalrows)
1039                                 stadistinct = totalrows;
1040                         stats->stadistinct = floor(stadistinct + 0.5);
1041                 }
1042
1043                 /*
1044                  * If we estimated the number of distinct values at more than 10%
1045                  * of the total row count (a very arbitrary limit), then assume
1046                  * that stadistinct should scale with the row count rather than be
1047                  * a fixed value.
1048                  */
1049                 if (stats->stadistinct > 0.1 * totalrows)
1050                         stats->stadistinct = -(stats->stadistinct / totalrows);
1051
1052                 /*
1053                  * Decide how many values are worth storing as most-common values.
1054                  * If we are able to generate a complete MCV list (all the values
1055                  * in the sample will fit, and we think these are all the ones in
1056                  * the table), then do so.      Otherwise, store only those values
1057                  * that are significantly more common than the (estimated)
1058                  * average. We set the threshold rather arbitrarily at 25% more
1059                  * than average, with at least 2 instances in the sample.
1060                  */
1061                 if (track_cnt < track_max && toowide_cnt == 0 &&
1062                         stats->stadistinct > 0 &&
1063                         track_cnt <= num_mcv)
1064                 {
1065                         /* Track list includes all values seen, and all will fit */
1066                         num_mcv = track_cnt;
1067                 }
1068                 else
1069                 {
1070                         double          ndistinct = stats->stadistinct;
1071                         double          avgcount,
1072                                                 mincount;
1073
1074                         if (ndistinct < 0)
1075                                 ndistinct = -ndistinct * totalrows;
1076                         /* estimate # of occurrences in sample of a typical value */
1077                         avgcount = (double) numrows / ndistinct;
1078                         /* set minimum threshold count to store a value */
1079                         mincount = avgcount * 1.25;
1080                         if (mincount < 2)
1081                                 mincount = 2;
1082                         if (num_mcv > track_cnt)
1083                                 num_mcv = track_cnt;
1084                         for (i = 0; i < num_mcv; i++)
1085                         {
1086                                 if (track[i].count < mincount)
1087                                 {
1088                                         num_mcv = i;
1089                                         break;
1090                                 }
1091                         }
1092                 }
1093
1094                 /* Generate MCV slot entry */
1095                 if (num_mcv > 0)
1096                 {
1097                         MemoryContext old_context;
1098                         Datum      *mcv_values;
1099                         float4     *mcv_freqs;
1100
1101                         /* Must copy the target values into TransactionCommandContext */
1102                         old_context = MemoryContextSwitchTo(TransactionCommandContext);
1103                         mcv_values = (Datum *) palloc(num_mcv * sizeof(Datum));
1104                         mcv_freqs = (float4 *) palloc(num_mcv * sizeof(float4));
1105                         for (i = 0; i < num_mcv; i++)
1106                         {
1107                                 mcv_values[i] = datumCopy(track[i].value,
1108                                                                                   stats->attr->attbyval,
1109                                                                                   stats->attr->attlen);
1110                                 mcv_freqs[i] = (double) track[i].count / (double) numrows;
1111                         }
1112                         MemoryContextSwitchTo(old_context);
1113
1114                         stats->stakind[0] = STATISTIC_KIND_MCV;
1115                         stats->staop[0] = stats->eqopr;
1116                         stats->stanumbers[0] = mcv_freqs;
1117                         stats->numnumbers[0] = num_mcv;
1118                         stats->stavalues[0] = mcv_values;
1119                         stats->numvalues[0] = num_mcv;
1120                 }
1121         }
1122
1123         /* We don't need to bother cleaning up any of our temporary palloc's */
1124 }
1125
1126
1127 /*
1128  *      compute_scalar_stats() -- compute column statistics
1129  *
1130  *      We use this when we can find "=" and "<" operators for the datatype.
1131  *
1132  *      We determine the fraction of non-null rows, the average width, the
1133  *      most common values, the (estimated) number of distinct values, the
1134  *      distribution histogram, and the correlation of physical to logical order.
1135  *
1136  *      The desired stats can be determined fairly easily after sorting the
1137  *      data values into order.
1138  */
1139 static void
1140 compute_scalar_stats(VacAttrStats *stats,
1141                                          TupleDesc tupDesc, double totalrows,
1142                                          HeapTuple *rows, int numrows)
1143 {
1144         int                     i;
1145         int                     null_cnt = 0;
1146         int                     nonnull_cnt = 0;
1147         int                     toowide_cnt = 0;
1148         double          total_width = 0;
1149         bool            is_varlena = (!stats->attr->attbyval &&
1150                                                           stats->attr->attlen == -1);
1151         double          corr_xysum;
1152         RegProcedure cmpFn;
1153         SortFunctionKind cmpFnKind;
1154         FmgrInfo        f_cmpfn;
1155         ScalarItem *values;
1156         int                     values_cnt = 0;
1157         int                *tupnoLink;
1158         ScalarMCVItem *track;
1159         int                     track_cnt = 0;
1160         int                     num_mcv = stats->attr->attstattarget;
1161         int                     num_bins = stats->attr->attstattarget;
1162
1163         values = (ScalarItem *) palloc(numrows * sizeof(ScalarItem));
1164         tupnoLink = (int *) palloc(numrows * sizeof(int));
1165         track = (ScalarMCVItem *) palloc(num_mcv * sizeof(ScalarMCVItem));
1166
1167         SelectSortFunction(stats->ltopr, &cmpFn, &cmpFnKind);
1168         fmgr_info(cmpFn, &f_cmpfn);
1169
1170         /* Initial scan to find sortable values */
1171         for (i = 0; i < numrows; i++)
1172         {
1173                 HeapTuple       tuple = rows[i];
1174                 Datum           value;
1175                 bool            isnull;
1176
1177                 CHECK_FOR_INTERRUPTS();
1178
1179                 value = heap_getattr(tuple, stats->attnum, tupDesc, &isnull);
1180
1181                 /* Check for null/nonnull */
1182                 if (isnull)
1183                 {
1184                         null_cnt++;
1185                         continue;
1186                 }
1187                 nonnull_cnt++;
1188
1189                 /*
1190                  * If it's a varlena field, add up widths for average width
1191                  * calculation.  Note that if the value is toasted, we use the
1192                  * toasted width.  We don't bother with this calculation if it's a
1193                  * fixed-width type.
1194                  */
1195                 if (is_varlena)
1196                 {
1197                         total_width += VARSIZE(DatumGetPointer(value));
1198
1199                         /*
1200                          * If the value is toasted, we want to detoast it just once to
1201                          * avoid repeated detoastings and resultant excess memory
1202                          * usage during the comparisons.  Also, check to see if the
1203                          * value is excessively wide, and if so don't detoast at all
1204                          * --- just ignore the value.
1205                          */
1206                         if (toast_raw_datum_size(value) > WIDTH_THRESHOLD)
1207                         {
1208                                 toowide_cnt++;
1209                                 continue;
1210                         }
1211                         value = PointerGetDatum(PG_DETOAST_DATUM(value));
1212                 }
1213
1214                 /* Add it to the list to be sorted */
1215                 values[values_cnt].value = value;
1216                 values[values_cnt].tupno = values_cnt;
1217                 tupnoLink[values_cnt] = values_cnt;
1218                 values_cnt++;
1219         }
1220
1221         /* We can only compute valid stats if we found some sortable values. */
1222         if (values_cnt > 0)
1223         {
1224                 int                     ndistinct,      /* # distinct values in sample */
1225                                         nmultiple,      /* # that appear multiple times */
1226                                         num_hist,
1227                                         dups_cnt;
1228                 int                     slot_idx = 0;
1229
1230                 /* Sort the collected values */
1231                 datumCmpFn = &f_cmpfn;
1232                 datumCmpFnKind = cmpFnKind;
1233                 datumCmpTupnoLink = tupnoLink;
1234                 qsort((void *) values, values_cnt,
1235                           sizeof(ScalarItem), compare_scalars);
1236
1237                 /*
1238                  * Now scan the values in order, find the most common ones, and
1239                  * also accumulate ordering-correlation statistics.
1240                  *
1241                  * To determine which are most common, we first have to count the
1242                  * number of duplicates of each value.  The duplicates are
1243                  * adjacent in the sorted list, so a brute-force approach is to
1244                  * compare successive datum values until we find two that are not
1245                  * equal. However, that requires N-1 invocations of the datum
1246                  * comparison routine, which are completely redundant with work
1247                  * that was done during the sort.  (The sort algorithm must at
1248                  * some point have compared each pair of items that are adjacent
1249                  * in the sorted order; otherwise it could not know that it's
1250                  * ordered the pair correctly.) We exploit this by having
1251                  * compare_scalars remember the highest tupno index that each
1252                  * ScalarItem has been found equal to.  At the end of the sort, a
1253                  * ScalarItem's tupnoLink will still point to itself if and only
1254                  * if it is the last item of its group of duplicates (since the
1255                  * group will be ordered by tupno).
1256                  */
1257                 corr_xysum = 0;
1258                 ndistinct = 0;
1259                 nmultiple = 0;
1260                 dups_cnt = 0;
1261                 for (i = 0; i < values_cnt; i++)
1262                 {
1263                         int                     tupno = values[i].tupno;
1264
1265                         corr_xysum += ((double) i) * ((double) tupno);
1266                         dups_cnt++;
1267                         if (tupnoLink[tupno] == tupno)
1268                         {
1269                                 /* Reached end of duplicates of this value */
1270                                 ndistinct++;
1271                                 if (dups_cnt > 1)
1272                                 {
1273                                         nmultiple++;
1274                                         if (track_cnt < num_mcv ||
1275                                                 dups_cnt > track[track_cnt - 1].count)
1276                                         {
1277                                                 /*
1278                                                  * Found a new item for the mcv list; find its
1279                                                  * position, bubbling down old items if needed.
1280                                                  * Loop invariant is that j points at an empty/
1281                                                  * replaceable slot.
1282                                                  */
1283                                                 int                     j;
1284
1285                                                 if (track_cnt < num_mcv)
1286                                                         track_cnt++;
1287                                                 for (j = track_cnt - 1; j > 0; j--)
1288                                                 {
1289                                                         if (dups_cnt <= track[j - 1].count)
1290                                                                 break;
1291                                                         track[j].count = track[j - 1].count;
1292                                                         track[j].first = track[j - 1].first;
1293                                                 }
1294                                                 track[j].count = dups_cnt;
1295                                                 track[j].first = i + 1 - dups_cnt;
1296                                         }
1297                                 }
1298                                 dups_cnt = 0;
1299                         }
1300                 }
1301
1302                 stats->stats_valid = true;
1303                 /* Do the simple null-frac and width stats */
1304                 stats->stanullfrac = (double) null_cnt / (double) numrows;
1305                 if (is_varlena)
1306                         stats->stawidth = total_width / (double) nonnull_cnt;
1307                 else
1308                         stats->stawidth = stats->attrtype->typlen;
1309
1310                 if (nmultiple == 0)
1311                 {
1312                         /* If we found no repeated values, assume it's a unique column */
1313                         stats->stadistinct = -1.0;
1314                 }
1315                 else if (toowide_cnt == 0 && nmultiple == ndistinct)
1316                 {
1317                         /*
1318                          * Every value in the sample appeared more than once.  Assume
1319                          * the column has just these values.
1320                          */
1321                         stats->stadistinct = ndistinct;
1322                 }
1323                 else
1324                 {
1325                         /*----------
1326                          * Estimate the number of distinct values using the estimator
1327                          * proposed by Haas and Stokes in IBM Research Report RJ 10025:
1328                          *              n*d / (n - f1 + f1*n/N)
1329                          * where f1 is the number of distinct values that occurred
1330                          * exactly once in our sample of n rows (from a total of N),
1331                          * and d is the total number of distinct values in the sample.
1332                          * This is their Duj1 estimator; the other estimators they
1333                          * recommend are considerably more complex, and are numerically
1334                          * very unstable when n is much smaller than N.
1335                          *
1336                          * Overwidth values are assumed to have been distinct.
1337                          *----------
1338                          */
1339                         int                     f1 = ndistinct - nmultiple + toowide_cnt;
1340                         int                     d = f1 + nmultiple;
1341                         double          numer, denom, stadistinct;
1342
1343                         numer = (double) numrows * (double) d;
1344                         denom = (double) (numrows - f1) +
1345                                 (double) f1 * (double) numrows / totalrows;
1346                         stadistinct = numer / denom;
1347                         /* Clamp to sane range in case of roundoff error */
1348                         if (stadistinct < (double) d)
1349                                 stadistinct = (double) d;
1350                         if (stadistinct > totalrows)
1351                                 stadistinct = totalrows;
1352                         stats->stadistinct = floor(stadistinct + 0.5);
1353                 }
1354
1355                 /*
1356                  * If we estimated the number of distinct values at more than 10%
1357                  * of the total row count (a very arbitrary limit), then assume
1358                  * that stadistinct should scale with the row count rather than be
1359                  * a fixed value.
1360                  */
1361                 if (stats->stadistinct > 0.1 * totalrows)
1362                         stats->stadistinct = -(stats->stadistinct / totalrows);
1363
1364                 /*
1365                  * Decide how many values are worth storing as most-common values.
1366                  * If we are able to generate a complete MCV list (all the values
1367                  * in the sample will fit, and we think these are all the ones in
1368                  * the table), then do so.      Otherwise, store only those values
1369                  * that are significantly more common than the (estimated)
1370                  * average. We set the threshold rather arbitrarily at 25% more
1371                  * than average, with at least 2 instances in the sample.  Also,
1372                  * we won't suppress values that have a frequency of at least 1/K
1373                  * where K is the intended number of histogram bins; such values
1374                  * might otherwise cause us to emit duplicate histogram bin
1375                  * boundaries.
1376                  */
1377                 if (track_cnt == ndistinct && toowide_cnt == 0 &&
1378                         stats->stadistinct > 0 &&
1379                         track_cnt <= num_mcv)
1380                 {
1381                         /* Track list includes all values seen, and all will fit */
1382                         num_mcv = track_cnt;
1383                 }
1384                 else
1385                 {
1386                         double          ndistinct = stats->stadistinct;
1387                         double          avgcount,
1388                                                 mincount,
1389                                                 maxmincount;
1390
1391                         if (ndistinct < 0)
1392                                 ndistinct = -ndistinct * totalrows;
1393                         /* estimate # of occurrences in sample of a typical value */
1394                         avgcount = (double) numrows / ndistinct;
1395                         /* set minimum threshold count to store a value */
1396                         mincount = avgcount * 1.25;
1397                         if (mincount < 2)
1398                                 mincount = 2;
1399                         /* don't let threshold exceed 1/K, however */
1400                         maxmincount = (double) numrows / (double) num_bins;
1401                         if (mincount > maxmincount)
1402                                 mincount = maxmincount;
1403                         if (num_mcv > track_cnt)
1404                                 num_mcv = track_cnt;
1405                         for (i = 0; i < num_mcv; i++)
1406                         {
1407                                 if (track[i].count < mincount)
1408                                 {
1409                                         num_mcv = i;
1410                                         break;
1411                                 }
1412                         }
1413                 }
1414
1415                 /* Generate MCV slot entry */
1416                 if (num_mcv > 0)
1417                 {
1418                         MemoryContext old_context;
1419                         Datum      *mcv_values;
1420                         float4     *mcv_freqs;
1421
1422                         /* Must copy the target values into TransactionCommandContext */
1423                         old_context = MemoryContextSwitchTo(TransactionCommandContext);
1424                         mcv_values = (Datum *) palloc(num_mcv * sizeof(Datum));
1425                         mcv_freqs = (float4 *) palloc(num_mcv * sizeof(float4));
1426                         for (i = 0; i < num_mcv; i++)
1427                         {
1428                                 mcv_values[i] = datumCopy(values[track[i].first].value,
1429                                                                                   stats->attr->attbyval,
1430                                                                                   stats->attr->attlen);
1431                                 mcv_freqs[i] = (double) track[i].count / (double) numrows;
1432                         }
1433                         MemoryContextSwitchTo(old_context);
1434
1435                         stats->stakind[slot_idx] = STATISTIC_KIND_MCV;
1436                         stats->staop[slot_idx] = stats->eqopr;
1437                         stats->stanumbers[slot_idx] = mcv_freqs;
1438                         stats->numnumbers[slot_idx] = num_mcv;
1439                         stats->stavalues[slot_idx] = mcv_values;
1440                         stats->numvalues[slot_idx] = num_mcv;
1441                         slot_idx++;
1442                 }
1443
1444                 /*
1445                  * Generate a histogram slot entry if there are at least two
1446                  * distinct values not accounted for in the MCV list.  (This
1447                  * ensures the histogram won't collapse to empty or a singleton.)
1448                  */
1449                 num_hist = ndistinct - num_mcv;
1450                 if (num_hist > num_bins)
1451                         num_hist = num_bins + 1;
1452                 if (num_hist >= 2)
1453                 {
1454                         MemoryContext old_context;
1455                         Datum      *hist_values;
1456                         int                     nvals;
1457
1458                         /* Sort the MCV items into position order to speed next loop */
1459                         qsort((void *) track, num_mcv,
1460                                   sizeof(ScalarMCVItem), compare_mcvs);
1461
1462                         /*
1463                          * Collapse out the MCV items from the values[] array.
1464                          *
1465                          * Note we destroy the values[] array here... but we don't need
1466                          * it for anything more.  We do, however, still need
1467                          * values_cnt. nvals will be the number of remaining entries
1468                          * in values[].
1469                          */
1470                         if (num_mcv > 0)
1471                         {
1472                                 int                     src,
1473                                                         dest;
1474                                 int                     j;
1475
1476                                 src = dest = 0;
1477                                 j = 0;                  /* index of next interesting MCV item */
1478                                 while (src < values_cnt)
1479                                 {
1480                                         int                     ncopy;
1481
1482                                         if (j < num_mcv)
1483                                         {
1484                                                 int                     first = track[j].first;
1485
1486                                                 if (src >= first)
1487                                                 {
1488                                                         /* advance past this MCV item */
1489                                                         src = first + track[j].count;
1490                                                         j++;
1491                                                         continue;
1492                                                 }
1493                                                 ncopy = first - src;
1494                                         }
1495                                         else
1496                                                 ncopy = values_cnt - src;
1497                                         memmove(&values[dest], &values[src],
1498                                                         ncopy * sizeof(ScalarItem));
1499                                         src += ncopy;
1500                                         dest += ncopy;
1501                                 }
1502                                 nvals = dest;
1503                         }
1504                         else
1505                                 nvals = values_cnt;
1506                         Assert(nvals >= num_hist);
1507
1508                         /* Must copy the target values into TransactionCommandContext */
1509                         old_context = MemoryContextSwitchTo(TransactionCommandContext);
1510                         hist_values = (Datum *) palloc(num_hist * sizeof(Datum));
1511                         for (i = 0; i < num_hist; i++)
1512                         {
1513                                 int                     pos;
1514
1515                                 pos = (i * (nvals - 1)) / (num_hist - 1);
1516                                 hist_values[i] = datumCopy(values[pos].value,
1517                                                                                    stats->attr->attbyval,
1518                                                                                    stats->attr->attlen);
1519                         }
1520                         MemoryContextSwitchTo(old_context);
1521
1522                         stats->stakind[slot_idx] = STATISTIC_KIND_HISTOGRAM;
1523                         stats->staop[slot_idx] = stats->ltopr;
1524                         stats->stavalues[slot_idx] = hist_values;
1525                         stats->numvalues[slot_idx] = num_hist;
1526                         slot_idx++;
1527                 }
1528
1529                 /* Generate a correlation entry if there are multiple values */
1530                 if (values_cnt > 1)
1531                 {
1532                         MemoryContext old_context;
1533                         float4     *corrs;
1534                         double          corr_xsum,
1535                                                 corr_x2sum;
1536
1537                         /* Must copy the target values into TransactionCommandContext */
1538                         old_context = MemoryContextSwitchTo(TransactionCommandContext);
1539                         corrs = (float4 *) palloc(sizeof(float4));
1540                         MemoryContextSwitchTo(old_context);
1541
1542                         /*----------
1543                          * Since we know the x and y value sets are both
1544                          *              0, 1, ..., values_cnt-1
1545                          * we have sum(x) = sum(y) =
1546                          *              (values_cnt-1)*values_cnt / 2
1547                          * and sum(x^2) = sum(y^2) =
1548                          *              (values_cnt-1)*values_cnt*(2*values_cnt-1) / 6.
1549                          *----------
1550                          */
1551                         corr_xsum = ((double) (values_cnt - 1)) *
1552                                 ((double) values_cnt) / 2.0;
1553                         corr_x2sum = ((double) (values_cnt - 1)) *
1554                                 ((double) values_cnt) * (double) (2 * values_cnt - 1) / 6.0;
1555
1556                         /* And the correlation coefficient reduces to */
1557                         corrs[0] = (values_cnt * corr_xysum - corr_xsum * corr_xsum) /
1558                                 (values_cnt * corr_x2sum - corr_xsum * corr_xsum);
1559
1560                         stats->stakind[slot_idx] = STATISTIC_KIND_CORRELATION;
1561                         stats->staop[slot_idx] = stats->ltopr;
1562                         stats->stanumbers[slot_idx] = corrs;
1563                         stats->numnumbers[slot_idx] = 1;
1564                         slot_idx++;
1565                 }
1566         }
1567
1568         /* We don't need to bother cleaning up any of our temporary palloc's */
1569 }
1570
1571 /*
1572  * qsort comparator for sorting ScalarItems
1573  *
1574  * Aside from sorting the items, we update the datumCmpTupnoLink[] array
1575  * whenever two ScalarItems are found to contain equal datums.  The array
1576  * is indexed by tupno; for each ScalarItem, it contains the highest
1577  * tupno that that item's datum has been found to be equal to.  This allows
1578  * us to avoid additional comparisons in compute_scalar_stats().
1579  */
1580 static int
1581 compare_scalars(const void *a, const void *b)
1582 {
1583         Datum           da = ((ScalarItem *) a)->value;
1584         int                     ta = ((ScalarItem *) a)->tupno;
1585         Datum           db = ((ScalarItem *) b)->value;
1586         int                     tb = ((ScalarItem *) b)->tupno;
1587         int32           compare;
1588
1589         compare = ApplySortFunction(datumCmpFn, datumCmpFnKind,
1590                                                                 da, false, db, false);
1591         if (compare != 0)
1592                 return compare;
1593
1594         /*
1595          * The two datums are equal, so update datumCmpTupnoLink[].
1596          */
1597         if (datumCmpTupnoLink[ta] < tb)
1598                 datumCmpTupnoLink[ta] = tb;
1599         if (datumCmpTupnoLink[tb] < ta)
1600                 datumCmpTupnoLink[tb] = ta;
1601
1602         /*
1603          * For equal datums, sort by tupno
1604          */
1605         return ta - tb;
1606 }
1607
1608 /*
1609  * qsort comparator for sorting ScalarMCVItems by position
1610  */
1611 static int
1612 compare_mcvs(const void *a, const void *b)
1613 {
1614         int                     da = ((ScalarMCVItem *) a)->first;
1615         int                     db = ((ScalarMCVItem *) b)->first;
1616
1617         return da - db;
1618 }
1619
1620
1621 /*
1622  *      update_attstats() -- update attribute statistics for one relation
1623  *
1624  *              Statistics are stored in several places: the pg_class row for the
1625  *              relation has stats about the whole relation, and there is a
1626  *              pg_statistic row for each (non-system) attribute that has ever
1627  *              been analyzed.  The pg_class values are updated by VACUUM, not here.
1628  *
1629  *              pg_statistic rows are just added or updated normally.  This means
1630  *              that pg_statistic will probably contain some deleted rows at the
1631  *              completion of a vacuum cycle, unless it happens to get vacuumed last.
1632  *
1633  *              To keep things simple, we punt for pg_statistic, and don't try
1634  *              to compute or store rows for pg_statistic itself in pg_statistic.
1635  *              This could possibly be made to work, but it's not worth the trouble.
1636  *              Note analyze_rel() has seen to it that we won't come here when
1637  *              vacuuming pg_statistic itself.
1638  */
1639 static void
1640 update_attstats(Oid relid, int natts, VacAttrStats **vacattrstats)
1641 {
1642         Relation        sd;
1643         int                     attno;
1644
1645         /*
1646          * We use an ExclusiveLock on pg_statistic to ensure that only one
1647          * backend is writing it at a time --- without that, we might have to
1648          * deal with concurrent updates here, and it's not worth the trouble.
1649          */
1650         sd = heap_openr(StatisticRelationName, ExclusiveLock);
1651
1652         for (attno = 0; attno < natts; attno++)
1653         {
1654                 VacAttrStats *stats = vacattrstats[attno];
1655                 FmgrInfo        out_function;
1656                 HeapTuple       stup,
1657                                         oldtup;
1658                 int                     i,
1659                                         k,
1660                                         n;
1661                 Datum           values[Natts_pg_statistic];
1662                 char            nulls[Natts_pg_statistic];
1663                 char            replaces[Natts_pg_statistic];
1664                 Relation        irelations[Num_pg_statistic_indices];
1665
1666                 /* Ignore attr if we weren't able to collect stats */
1667                 if (!stats->stats_valid)
1668                         continue;
1669
1670                 fmgr_info(stats->attrtype->typoutput, &out_function);
1671
1672                 /*
1673                  * Construct a new pg_statistic tuple
1674                  */
1675                 for (i = 0; i < Natts_pg_statistic; ++i)
1676                 {
1677                         nulls[i] = ' ';
1678                         replaces[i] = 'r';
1679                 }
1680
1681                 i = 0;
1682                 values[i++] = ObjectIdGetDatum(relid);  /* starelid */
1683                 values[i++] = Int16GetDatum(stats->attnum);             /* staattnum */
1684                 values[i++] = Float4GetDatum(stats->stanullfrac);               /* stanullfrac */
1685                 values[i++] = Int32GetDatum(stats->stawidth);   /* stawidth */
1686                 values[i++] = Float4GetDatum(stats->stadistinct);               /* stadistinct */
1687                 for (k = 0; k < STATISTIC_NUM_SLOTS; k++)
1688                 {
1689                         values[i++] = Int16GetDatum(stats->stakind[k]);         /* stakindN */
1690                 }
1691                 for (k = 0; k < STATISTIC_NUM_SLOTS; k++)
1692                 {
1693                         values[i++] = ObjectIdGetDatum(stats->staop[k]);        /* staopN */
1694                 }
1695                 for (k = 0; k < STATISTIC_NUM_SLOTS; k++)
1696                 {
1697                         int                     nnum = stats->numnumbers[k];
1698
1699                         if (nnum > 0)
1700                         {
1701                                 Datum      *numdatums = (Datum *) palloc(nnum * sizeof(Datum));
1702                                 ArrayType  *arry;
1703
1704                                 for (n = 0; n < nnum; n++)
1705                                         numdatums[n] = Float4GetDatum(stats->stanumbers[k][n]);
1706                                 /* XXX knows more than it should about type float4: */
1707                                 arry = construct_array(numdatums, nnum,
1708                                                                            false, sizeof(float4), 'i');
1709                                 values[i++] = PointerGetDatum(arry);    /* stanumbersN */
1710                         }
1711                         else
1712                         {
1713                                 nulls[i] = 'n';
1714                                 values[i++] = (Datum) 0;
1715                         }
1716                 }
1717                 for (k = 0; k < STATISTIC_NUM_SLOTS; k++)
1718                 {
1719                         int                     ntxt = stats->numvalues[k];
1720
1721                         if (ntxt > 0)
1722                         {
1723                                 Datum      *txtdatums = (Datum *) palloc(ntxt * sizeof(Datum));
1724                                 ArrayType  *arry;
1725
1726                                 for (n = 0; n < ntxt; n++)
1727                                 {
1728                                         /*
1729                                          * Convert data values to a text string to be inserted
1730                                          * into the text array.
1731                                          */
1732                                         Datum           stringdatum;
1733
1734                                         stringdatum =
1735                                                 FunctionCall3(&out_function,
1736                                                                           stats->stavalues[k][n],
1737                                                           ObjectIdGetDatum(stats->attrtype->typelem),
1738                                                                   Int32GetDatum(stats->attr->atttypmod));
1739                                         txtdatums[n] = DirectFunctionCall1(textin, stringdatum);
1740                                         pfree(DatumGetPointer(stringdatum));
1741                                 }
1742                                 /* XXX knows more than it should about type text: */
1743                                 arry = construct_array(txtdatums, ntxt,
1744                                                                            false, -1, 'i');
1745                                 values[i++] = PointerGetDatum(arry);    /* stavaluesN */
1746                         }
1747                         else
1748                         {
1749                                 nulls[i] = 'n';
1750                                 values[i++] = (Datum) 0;
1751                         }
1752                 }
1753
1754                 /* Is there already a pg_statistic tuple for this attribute? */
1755                 oldtup = SearchSysCache(STATRELATT,
1756                                                                 ObjectIdGetDatum(relid),
1757                                                                 Int16GetDatum(stats->attnum),
1758                                                                 0, 0);
1759
1760                 if (HeapTupleIsValid(oldtup))
1761                 {
1762                         /* Yes, replace it */
1763                         stup = heap_modifytuple(oldtup,
1764                                                                         sd,
1765                                                                         values,
1766                                                                         nulls,
1767                                                                         replaces);
1768                         ReleaseSysCache(oldtup);
1769                         simple_heap_update(sd, &stup->t_self, stup);
1770                 }
1771                 else
1772                 {
1773                         /* No, insert new tuple */
1774                         stup = heap_formtuple(sd->rd_att, values, nulls);
1775                         heap_insert(sd, stup);
1776                 }
1777
1778                 /* update indices too */
1779                 CatalogOpenIndices(Num_pg_statistic_indices, Name_pg_statistic_indices,
1780                                                    irelations);
1781                 CatalogIndexInsert(irelations, Num_pg_statistic_indices, sd, stup);
1782                 CatalogCloseIndices(Num_pg_statistic_indices, irelations);
1783
1784                 heap_freetuple(stup);
1785         }
1786
1787         /* close rel, but hold lock till upcoming commit */
1788         heap_close(sd, NoLock);
1789 }