src/backend/commands/analyze.c

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