get correlation from segments instead of calculating it on QD (#15357)

We cannot use the same method as PostgreSQL does to calculate the correlation in QD.
When we collect data from segments to QD, this will change the physical order of the data.
such as in segment 1 the data is 1,3,5,7,9. And in segment 2 the data is 2,4,6,8,10.
In each segment the data is ordered, and correlation is 1 in each segment.
But after we collect the data to QD, it may be 1,3,5,2,4,7,9,6,8,10. And the correlation
is 0.3 or something else and it is not stable.
And this will increase the cost of index scan which is shouldn't be done.

So get correlations from segments and then calculate correlation for QD.

we use the weighted mean algorithm to calculate correlation on QD,
However, In some situations, we may not be able to obtain reltuples of a table,
such as none-leaf part of partitioned table or the parent table of the
inherited table. So we can only use the mean algorithm to calculate correlation for these tables.

Author: zxuejing

src/backend/commands/analyze.c

@@ -29,6 +29,7 @@
 #include "foreign/fdwapi.h"
 #include "miscadmin.h"
 #include "funcapi.h"
+#include "utils/syscache.h"
 
 /**
  * Statistics related parameters.
@@ -368,3 +369,159 @@ gp_acquire_sample_rows_col_type(Oid typid)
 	}
 	return typid;
 }
+
+/*
+ * gp_acquire_correlations - Acquire each column's correlation for a table.
+ * This is an internal function called in gp_acquire_correlations_dispatcher.
+ * this function will return a result set, a row for each alive column.
+ * each row contains 3 columns: attnum, the correlation for it and totalrows.
+ * if correlation is null, set totalrows to 0 for it.
+ *
+ * So overall, this returns a result set like this:
+ * create table t(tc1 int, tc2 int, tc3 int);
+ * insert values.
+ * alter table t drop column tc2;
+ *
+ *    attnum | correlation| totalrows
+ * ----------+------------|+------------
+ *      0    |      0.8   | 200
+ *      2    |            | 0
+ */
+Datum
+gp_acquire_correlations(PG_FUNCTION_ARGS)
+{
+	FuncCallContext *funcctx = NULL;
+	gp_acquire_correlation_context *ctx;
+	MemoryContext oldcontext;
+	Oid			relOid = PG_GETARG_OID(0);
+	bool		inherited = PG_GETARG_BOOL(1);
+	TupleDesc	relDesc;
+	TupleDesc	outDesc;
+
+	if (SRF_IS_FIRSTCALL())
+	{
+		Relation	onerel;
+		funcctx = SRF_FIRSTCALL_INIT();
+
+		/*
+		 * switch to memory context appropriate for multiple function
+		 * calls
+		 */
+		oldcontext = MemoryContextSwitchTo(funcctx->multi_call_memory_ctx);
+
+		/* Construct the context to keep across calls. */
+		ctx = (gp_acquire_correlation_context *) palloc0(sizeof(gp_acquire_correlation_context));
+
+		if (!pg_class_ownercheck(relOid, GetUserId()))
+			aclcheck_error(ACLCHECK_NOT_OWNER, OBJECT_TABLE,
+						   get_rel_name(relOid));
+
+		onerel = table_open(relOid, AccessShareLock);
+		relDesc = RelationGetDescr(onerel);
+
+		outDesc = CreateTemplateTupleDesc(3);
+		TupleDescInitEntry(outDesc,
+						   1,
+						   "attnum",
+						   INT4OID,
+						   -1,
+						   0);
+		TupleDescInitEntry(outDesc,
+						   2,
+						   "correlation",
+						   FLOAT4OID,
+						   -1,
+						   0);
+		TupleDescInitEntry(outDesc,
+						   3,
+						   "totalrows",
+						   INT4OID,
+						   -1,
+						   0);
+
+		BlessTupleDesc(outDesc);
+		funcctx->tuple_desc = outDesc;
+
+		ctx->onerel = onerel;
+		funcctx->user_fctx = ctx;
+		ctx->outDesc = outDesc;
+
+		ctx->index = 0;
+		ctx->totalAttr = relDesc->natts;
+		MemoryContextSwitchTo(oldcontext);
+	}
+
+	/* stuff done on every call of the function */
+	funcctx = SRF_PERCALL_SETUP();
+
+	ctx = funcctx->user_fctx;
+	relDesc = RelationGetDescr(ctx->onerel);
+	outDesc = ctx->outDesc;
+
+	Datum		*outvalues = (Datum *) palloc(outDesc->natts * sizeof(Datum));
+	bool		*outnulls = (bool *) palloc(outDesc->natts * sizeof(bool));
+	HeapTuple	res;
+	int attno = ctx->index;
+
+	/* Return all alive attribute correlation */
+	for (; attno < ctx->totalAttr; attno++)
+	{
+		/* get the correlation of the column */
+		int			totalrows = 0;
+		HeapTuple	statsTuple;
+		Form_pg_attribute relatt = TupleDescAttr(relDesc, attno);
+		if (relatt->attisdropped)
+			continue;
+		statsTuple = SearchSysCache3(STATRELATTINH,
+												ObjectIdGetDatum(relOid),
+												Int16GetDatum(attno + 1),
+												BoolGetDatum(inherited));
+		outvalues[0] = Int32GetDatum(attno);
+		outnulls[0] = false;
+
+		if (HeapTupleIsValid(statsTuple))
+		{
+			AttStatsSlot sslot;
+
+			if (get_attstatsslot(&sslot, statsTuple,
+						 STATISTIC_KIND_CORRELATION, InvalidOid,
+						 ATTSTATSSLOT_NUMBERS))
+			{
+				float4		varCorrelation;
+				Assert(sslot.nnumbers == 1);
+				varCorrelation = sslot.numbers[0];
+
+				free_attstatsslot(&sslot);
+
+				outvalues[1] = Float4GetDatum(varCorrelation);
+				outnulls[1] = false;
+				totalrows = ctx->onerel->rd_rel->reltuples;
+			}
+			else
+			{
+				outvalues[1] = (Datum) 0;
+				outnulls[1] = true;
+			}
+			ReleaseSysCache(statsTuple);
+		}
+		else
+		{
+			outvalues[1] = (Datum) 0;
+			outnulls[1] = true;
+		}
+
+		outvalues[2] = Int32GetDatum(totalrows);
+		outnulls[2] = false;
+
+		res = heap_form_tuple(outDesc, outvalues, outnulls);
+		ctx->index = attno + 1;
+
+		SRF_RETURN_NEXT(funcctx, HeapTupleGetDatum(res));
+	}
+
+	table_close(ctx->onerel, AccessShareLock);
+	pfree(ctx);
+	funcctx->user_fctx = NULL;
+
+	SRF_RETURN_DONE(funcctx);
+}

src/backend/commands/analyzefuncs.c

@@ -11810,7 +11810,7 @@
 { oid => 6464, descr => 'get backends of overflowed subtransaction',
    proname => 'gp_get_suboverflowed_backends', provolatile => 'v', prorettype => '_int4', proargtypes => '', prosrc => 'gp_get_suboverflowed_backends' },
 
-{ oid => 6040, descr => 'get gp all segments pg_snapshot',
+{ oid => 6041, descr => 'get gp all segments pg_snapshot',
    proname => 'gp_current_snapshot', proisstrict => 'f',
    proretset => 't', provolatile => 'v', proparallel => 'r',
    prorettype => 'record', proargtypes => '',
@@ -11992,6 +11992,8 @@
 # Analyze related
 { oid => 6038, descr => 'Collect a random sample of rows from table',
    proname => 'gp_acquire_sample_rows', prorows => '1000', proretset => 't', provolatile => 'v', proparallel => 'u', prorettype => 'record', proargtypes => 'oid int4 bool', prosrc => 'gp_acquire_sample_rows', proexeclocation => 's' },
+{ oid => 6040, descr => 'Collect correlations from segments',
+   proname => 'gp_acquire_correlations', prorows => '10', proretset => 't', provolatile => 'v', proparallel => 'u', prorettype => 'record', proargtypes => 'oid bool', prosrc => 'gp_acquire_correlations', proexeclocation => 's' },
 
 # Backoff related
 { oid => 7016, descr => 'change weight of all the backends for a given session id',

src/include/catalog/pg_proc.dat

@@ -179,6 +179,9 @@ typedef struct VacAttrStats
 	bool	   *exprnulls;
 	int			rowstride;
 	bool		merge_stats;
+	bool		corrnull;	  /* whether correlation value is null */
+	bool		partitiontbl_qd; /* analyze is on QD and the policy of table is partitioned */
+	float4		corrval;	  /* correlation gathered from segments */
 } VacAttrStats;
 
 
@@ -327,6 +330,24 @@ typedef struct
 	bool		summary_sent;
 } gp_acquire_sample_rows_context;
 
+typedef struct
+{
+	/* Table being analyzed */
+	Relation	onerel;
+
+	/* whether acquire inherited table's correlations */
+	bool        inherited;
+
+	/*
+	 * Result tuple descriptor.
+	 */
+	TupleDesc	outDesc;
+
+	/* SRF state, to track which rows have already been returned. */
+	int			index;
+	int			totalAttr;
+} gp_acquire_correlation_context;
+
 /* GUC parameters */
 extern PGDLLIMPORT int default_statistics_target;	/* PGDLLIMPORT for PostGIS */
 extern int	vacuum_freeze_min_age;
@@ -416,6 +437,7 @@ extern int acquire_inherited_sample_rows(Relation onerel, int elevel,
 
 /* in commands/analyzefuncs.c */
 extern Datum gp_acquire_sample_rows(PG_FUNCTION_ARGS);
+extern Datum gp_acquire_correlations(PG_FUNCTION_ARGS);
 extern Oid gp_acquire_sample_rows_col_type(Oid typid);
 
 extern bool gp_vacuum_needs_update_stats(void);