ORCA: allow different strategy control the redistribute key below aggregate

In CBDB, if there is an AGG operator (one-step AGG or final AGG operator) that
requires data redistribution, then the redistribution motion operator will used
all `GROUP BY` keys as the redistribute keys. In fact, only a single key needs
to be redistributed, and the results of AGG will be the same.

Reducing the number of redistributed keys can effectively reduce the overhead
of hash function calls in motion operator. However, this may lead to data skew.

Therefore, the current commit provides several different strategies for deciding
how redistribution keys should be selected during redistribution motion operator
(which under the AGG operator). User can use the GUC `optimizer_agg_pds_strategy`
to select the strategies.

- OPTIMIZER_AGG_PDS_ALL_KEY(value: 0): default one, select all `GROUP BY` key as
  the redistributed keys.
- OPTIMIZER_AGG_PDS_FIRST_KEY(value: 1): select the first `GROUP BY` key as the
  redistributed keys.
- OPTIMIZER_AGG_PDS_MINIMAL_LEN_KEY(value: 2): select a `GROUP BY` key which has
  the minimal and positive typlen as the redistributed keys. If only non-fixed
  type (such as text and varchar) exist, select the first `GROUP BY` key.
- OPTIMIZER_AGG_PDS_EXCLUDE_NON_FIXED(value: 3): select the `GROUP BY` key which
  is fixed typlen the redistributed keys. If only non-fixed type (such as text
  and varchar) exist, select the first `GROUP BY` key.

Author: jiaqizho

src/backend/gpopt/config/CConfigParamMapping.cpp

@@ -578,6 +578,14 @@ CConfigParamMapping::PackConfigParamInBitset(
 		traceflag_bitset->ExchangeSet(EopttraceEnableWindowHashAgg);
 	}
 
+	if (optimizer_agg_pds_strategy == OPTIMIZER_AGG_PDS_FIRST_KEY) {
+		traceflag_bitset->ExchangeSet(EopttraceAggRRSFirstKey);
+	} else if (optimizer_agg_pds_strategy == OPTIMIZER_AGG_PDS_MINIMAL_LEN_KEY) {
+		traceflag_bitset->ExchangeSet(EopttraceAggRRSMinimalLenKey);
+	} else if (optimizer_agg_pds_strategy == OPTIMIZER_AGG_PDS_EXCLUDE_NON_FIXED) {
+		traceflag_bitset->ExchangeSet(EopttraceAggRRSExcludeNonFixedKey);
+	}
+
 	return traceflag_bitset;
 }
 

src/backend/gporca/libgpopt/src/operators/CPhysicalAgg.cpp

@@ -8,7 +8,6 @@
 //	@doc:
 //		Implementation of basic aggregate operator
 //---------------------------------------------------------------------------
-
 #include "gpopt/operators/CPhysicalAgg.h"
 
 #include "gpos/base.h"
@@ -295,11 +294,65 @@ CDistributionSpec *
 CPhysicalAgg::PdsMaximalHashed(CMemoryPool *mp, CColRefArray *colref_array)
 {
 	GPOS_ASSERT(nullptr != colref_array);
+	CColRefArray *pcraResHashs = nullptr;
+
+	if (GPOS_FTRACE(EopttraceAggRRSFirstKey)) {
+		pcraResHashs = GPOS_NEW(mp) CColRefArray(mp);
+		if (colref_array->Size() > 0) {
+			CColRef *colref = (*colref_array)[0];
+			pcraResHashs->Append(colref);
+		}
+	} else if (GPOS_FTRACE(EopttraceAggRRSMinimalLenKey)) {
+		ULONG ulsz = colref_array->Size();
+		pcraResHashs = GPOS_NEW(mp) CColRefArray(mp);
+
+		LINT minimal_typlen = gpos::lint_max; // less than minimal typlen
+		LINT minimal_typlen_ul = -1;
+		for (ULONG ul = 0; ul < ulsz; ul++) {
+			CColRef *pcr =(*colref_array)[ul];
+			const gpmd::IMDType *pmdtyp = pcr->RetrieveType();
+			LINT typlen = pmdtyp->IsFixedLength() ? pmdtyp->Length() : (gpos::lint_max - 1);
+
+			if (typlen < minimal_typlen) {
+				minimal_typlen = typlen;
+				minimal_typlen_ul = ul;
+			}
+		}
+
+		if (minimal_typlen_ul != -1) {
+			pcraResHashs->Append((*colref_array)[minimal_typlen_ul]);
+		}
+	} else if (GPOS_FTRACE(EopttraceAggRRSExcludeNonFixedKey)) {
+		ULONG ulsz = colref_array->Size();
+		pcraResHashs = GPOS_NEW(mp) CColRefArray(mp);
+
+		for (ULONG ul = 0; ul < ulsz; ul++) {
+			CColRef *pcr =(*colref_array)[ul];
+			const gpmd::IMDType *pmdtyp = pcr->RetrieveType();
+			if (pmdtyp->IsFixedLength()) {
+				pcraResHashs->Append(pcr);
+			}
+		}
+
+		// no key in result
+		if (pcraResHashs->Size() == 0) {
+			colref_array->AddRef();
+			pcraResHashs = colref_array;
+		}
+	} else {
+		colref_array->AddRef();
+		pcraResHashs = colref_array;
+	}
+
+	GPOS_ASSERT(nullptr != pcraResHashs);
+	GPOS_ASSERT_IMP(colref_array->Size() > 0, pcraResHashs->Size() > 0);
 
 	CDistributionSpecHashed *pdshashedMaximal =
 		CDistributionSpecHashed::PdshashedMaximal(
-			mp, colref_array, true /*fNullsColocated*/
+			mp, pcraResHashs, true /*fNullsColocated*/
 		);
+
+	pcraResHashs->Release();
 	if (nullptr != pdshashedMaximal)
 	{
 		return pdshashedMaximal;

src/backend/gporca/libnaucrates/include/naucrates/traceflags/traceflags.h

@@ -244,6 +244,15 @@ enum EOptTraceFlag
 	// Enable window hash agg
 	EopttraceEnableWindowHashAgg = 103050,
 
+	// Use the first key in AGG Pds
+	EopttraceAggRRSFirstKey = 103051,
+
+	// Use the minimal length key in AGG Pds
+	EopttraceAggRRSMinimalLenKey = 103052,
+
+	// Use the all key exclude the non-fixed key in AGG pds
+	EopttraceAggRRSExcludeNonFixedKey = 103053,
+
 	///////////////////////////////////////////////////////
 	///////////////////// statistics flags ////////////////
 	//////////////////////////////////////////////////////

src/backend/utils/misc/guc_gp.c

@@ -365,6 +365,7 @@ bool		optimizer_enable_foreign_table;
 bool		optimizer_enable_right_outer_join;
 bool		optimizer_enable_query_parameter;
 bool		optimizer_force_window_hash_agg;
+int			optimizer_agg_pds_strategy;
 
 /* Optimizer plan enumeration related GUCs */
 bool		optimizer_enumerate_plans;
@@ -4490,6 +4491,17 @@ struct config_int ConfigureNamesInt_gp[] =
 		NULL, NULL, NULL
 	},
 
+	{
+		{"optimizer_agg_pds_strategy", PGC_USERSET, DEVELOPER_OPTIONS,
+			gettext_noop("Set the strategy of agg required distribution."),
+			NULL,
+			GUC_NOT_IN_SAMPLE
+		},
+		&optimizer_agg_pds_strategy,
+		OPTIMIZER_AGG_PDS_ALL_KEY, OPTIMIZER_AGG_PDS_ALL_KEY, OPTIMIZER_AGG_PDS_EXCLUDE_NON_FIXED,
+		NULL, NULL, NULL
+	},
+
 	{
 		{"memory_profiler_dataset_size", PGC_USERSET, DEVELOPER_OPTIONS,
 			gettext_noop("Set the size in GB"),

src/include/utils/guc.h

@@ -466,6 +466,12 @@ extern bool create_restartpoint_on_ckpt_record_replay;
 #define OPTIMIZER_GPDB_CALIBRATED       1       /* GPDB's calibrated cost model */
 #define OPTIMIZER_GPDB_EXPERIMENTAL     2       /* GPDB's experimental cost model */
 
+/* optimizer cost model */
+#define OPTIMIZER_AGG_PDS_ALL_KEY             0
+#define OPTIMIZER_AGG_PDS_FIRST_KEY           1
+#define OPTIMIZER_AGG_PDS_MINIMAL_LEN_KEY     2
+#define OPTIMIZER_AGG_PDS_EXCLUDE_NON_FIXED   3
+
 
 /* Optimizer related gucs */
 extern bool	optimizer;
@@ -546,6 +552,7 @@ extern bool optimizer_enable_foreign_table;
 extern bool optimizer_enable_right_outer_join;
 extern bool optimizer_enable_query_parameter;
 extern bool optimizer_force_window_hash_agg;
+extern int optimizer_agg_pds_strategy;
 
 /* Optimizer plan enumeration related GUCs */
 extern bool optimizer_enumerate_plans;

src/include/utils/unsync_guc_name.h

@@ -440,6 +440,7 @@
 		"optimizer_enable_right_outer_join",
 		"optimizer_enable_query_parameter",
 		"optimizer_force_window_hash_agg",
+		"optimizer_agg_pds_strategy",
 		"optimizer_enforce_subplans",
 		"optimizer_enumerate_plans",
 		"optimizer_expand_fulljoin",

src/test/regress/expected/aggregates.out

@@ -3404,3 +3404,93 @@ having sum(tgb1.v3 * tgb2.v3) > 100 and
 
 reset debug_print_aggref_in_explain;
 reset optimizer_force_multistage_agg;
+-- test the optimizer_agg_pds_strategy
+DROP TABLE IF EXISTS pds_t1;
+NOTICE:  table "pds_t1" does not exist, skipping
+create table pds_t1(v1 int, v2 text, v3 int, v4 text, v5 int, v6 text) DISTRIBUTED BY (v1);
+set optimizer_agg_pds_strategy to 0;
+explain (costs off) select v2,v3 from pds_t1 group by v2,v3;
+                         QUERY PLAN                         
+------------------------------------------------------------
+ Gather Motion 3:1  (slice1; segments: 3)
+   ->  HashAggregate
+         Group Key: v2, v3
+         ->  Redistribute Motion 3:3  (slice2; segments: 3)
+               Hash Key: v2, v3
+               ->  HashAggregate
+                     Group Key: v2, v3
+                     ->  Seq Scan on pds_t1
+ Optimizer: Postgres query optimizer
+(9 rows)
+
+set optimizer_agg_pds_strategy to 1;
+explain (costs off) select v2,v3 from pds_t1 group by v2,v3;
+                         QUERY PLAN                         
+------------------------------------------------------------
+ Gather Motion 3:1  (slice1; segments: 3)
+   ->  HashAggregate
+         Group Key: v2, v3
+         ->  Redistribute Motion 3:3  (slice2; segments: 3)
+               Hash Key: v2, v3
+               ->  HashAggregate
+                     Group Key: v2, v3
+                     ->  Seq Scan on pds_t1
+ Optimizer: Postgres query optimizer
+(9 rows)
+
+set optimizer_agg_pds_strategy to 2;
+explain (costs off) select v2,v3 from pds_t1 group by v2,v3;
+                         QUERY PLAN                         
+------------------------------------------------------------
+ Gather Motion 3:1  (slice1; segments: 3)
+   ->  HashAggregate
+         Group Key: v2, v3
+         ->  Redistribute Motion 3:3  (slice2; segments: 3)
+               Hash Key: v2, v3
+               ->  HashAggregate
+                     Group Key: v2, v3
+                     ->  Seq Scan on pds_t1
+ Optimizer: Postgres query optimizer
+(9 rows)
+
+set optimizer_agg_pds_strategy to 3;
+explain (costs off) select v2,v3,v4,v5,v6 from pds_t1 group by v2,v3,v4,v5,v6;
+                         QUERY PLAN                         
+------------------------------------------------------------
+ Gather Motion 3:1  (slice1; segments: 3)
+   ->  HashAggregate
+         Group Key: v2, v3, v4, v5, v6
+         ->  Redistribute Motion 3:3  (slice2; segments: 3)
+               Hash Key: v2, v3, v4, v5, v6
+               ->  HashAggregate
+                     Group Key: v2, v3, v4, v5, v6
+                     ->  Seq Scan on pds_t1
+ Optimizer: Postgres query optimizer
+(9 rows)
+
+-- We can't dedup the "Redistribute Motion", cause in this step we can't know the 
+-- distribution of output column which from the underlying operators.
+-- So you need to be cautious when opening this guc.
+set optimizer_agg_pds_strategy to 1;
+explain (costs off) select v1,v2,v3 from pds_t1 group by v3,v2,v1;
+                QUERY PLAN                
+------------------------------------------
+ Gather Motion 3:1  (slice1; segments: 3)
+   ->  HashAggregate
+         Group Key: v3, v2, v1
+         ->  Seq Scan on pds_t1
+ Optimizer: Postgres query optimizer
+(5 rows)
+
+set optimizer_agg_pds_strategy to 0;
+explain (costs off) select v1,v2,v3 from pds_t1 group by v3,v2,v1;
+                QUERY PLAN                
+------------------------------------------
+ Gather Motion 3:1  (slice1; segments: 3)
+   ->  HashAggregate
+         Group Key: v3, v2, v1
+         ->  Seq Scan on pds_t1
+ Optimizer: Postgres query optimizer
+(5 rows)
+
+reset optimizer_agg_pds_strategy;

src/test/regress/expected/aggregates_optimizer.out

@@ -3573,3 +3573,99 @@ having sum(tgb1.v3 * tgb2.v3) > 100 and
 
 reset debug_print_aggref_in_explain;
 reset optimizer_force_multistage_agg;
+-- test the optimizer_agg_pds_strategy
+DROP TABLE IF EXISTS pds_t1;
+NOTICE:  table "pds_t1" does not exist, skipping
+create table pds_t1(v1 int, v2 text, v3 int, v4 text, v5 int, v6 text) DISTRIBUTED BY (v1);
+set optimizer_agg_pds_strategy to 0;
+explain (costs off) select v2,v3 from pds_t1 group by v2,v3;
+                            QUERY PLAN                            
+------------------------------------------------------------------
+ Gather Motion 3:1  (slice1; segments: 3)
+   ->  GroupAggregate
+         Group Key: v2, v3
+         ->  Sort
+               Sort Key: v2, v3
+               ->  Redistribute Motion 3:3  (slice2; segments: 3)
+                     Hash Key: v2, v3
+                     ->  Seq Scan on pds_t1
+ Optimizer: GPORCA
+(9 rows)
+
+set optimizer_agg_pds_strategy to 1;
+explain (costs off) select v2,v3 from pds_t1 group by v2,v3;
+                            QUERY PLAN                            
+------------------------------------------------------------------
+ Gather Motion 3:1  (slice1; segments: 3)
+   ->  GroupAggregate
+         Group Key: v2, v3
+         ->  Sort
+               Sort Key: v2, v3
+               ->  Redistribute Motion 3:3  (slice2; segments: 3)
+                     Hash Key: v2
+                     ->  Seq Scan on pds_t1
+ Optimizer: GPORCA
+(9 rows)
+
+set optimizer_agg_pds_strategy to 2;
+explain (costs off) select v2,v3 from pds_t1 group by v2,v3;
+                            QUERY PLAN                            
+------------------------------------------------------------------
+ Gather Motion 3:1  (slice1; segments: 3)
+   ->  GroupAggregate
+         Group Key: v2, v3
+         ->  Sort
+               Sort Key: v2, v3
+               ->  Redistribute Motion 3:3  (slice2; segments: 3)
+                     Hash Key: v3
+                     ->  Seq Scan on pds_t1
+ Optimizer: GPORCA
+(9 rows)
+
+set optimizer_agg_pds_strategy to 3;
+explain (costs off) select v2,v3,v4,v5,v6 from pds_t1 group by v2,v3,v4,v5,v6;
+                            QUERY PLAN                            
+------------------------------------------------------------------
+ Gather Motion 3:1  (slice1; segments: 3)
+   ->  GroupAggregate
+         Group Key: v2, v3, v4, v5, v6
+         ->  Sort
+               Sort Key: v2, v3, v4, v5, v6
+               ->  Redistribute Motion 3:3  (slice2; segments: 3)
+                     Hash Key: v3, v5
+                     ->  Seq Scan on pds_t1
+ Optimizer: GPORCA
+(9 rows)
+
+-- We can't dedup the "Redistribute Motion", cause in this step we can't know the 
+-- distribution of output column which from the underlying operators.
+-- So you need to be cautious when opening this guc.
+set optimizer_agg_pds_strategy to 1;
+explain (costs off) select v1,v2,v3 from pds_t1 group by v3,v2,v1;
+                            QUERY PLAN                            
+------------------------------------------------------------------
+ Gather Motion 3:1  (slice1; segments: 3)
+   ->  GroupAggregate
+         Group Key: v3, v2, v1
+         ->  Sort
+               Sort Key: v3, v2, v1
+               ->  Redistribute Motion 3:3  (slice2; segments: 3)
+                     Hash Key: v3
+                     ->  Seq Scan on pds_t1
+ Optimizer: GPORCA
+(9 rows)
+
+set optimizer_agg_pds_strategy to 0;
+explain (costs off) select v1,v2,v3 from pds_t1 group by v3,v2,v1;
+                QUERY PLAN                
+------------------------------------------
+ Gather Motion 3:1  (slice1; segments: 3)
+   ->  GroupAggregate
+         Group Key: v3, v2, v1
+         ->  Sort
+               Sort Key: v3, v2, v1
+               ->  Seq Scan on pds_t1
+ Optimizer: GPORCA
+(7 rows)
+
+reset optimizer_agg_pds_strategy;

src/test/regress/sql/aggregates.sql

@@ -1528,3 +1528,26 @@ having sum(tgb1.v3 * tgb2.v3) > 100 and
 
 reset debug_print_aggref_in_explain;
 reset optimizer_force_multistage_agg;
+
+-- test the optimizer_agg_pds_strategy
+DROP TABLE IF EXISTS pds_t1;
+create table pds_t1(v1 int, v2 text, v3 int, v4 text, v5 int, v6 text) DISTRIBUTED BY (v1);
+
+set optimizer_agg_pds_strategy to 0;
+explain (costs off) select v2,v3 from pds_t1 group by v2,v3;
+set optimizer_agg_pds_strategy to 1;
+explain (costs off) select v2,v3 from pds_t1 group by v2,v3;
+set optimizer_agg_pds_strategy to 2;
+explain (costs off) select v2,v3 from pds_t1 group by v2,v3;
+set optimizer_agg_pds_strategy to 3;
+explain (costs off) select v2,v3,v4,v5,v6 from pds_t1 group by v2,v3,v4,v5,v6;
+
+-- We can't dedup the "Redistribute Motion", cause in this step we can't know the 
+-- distribution of output column which from the underlying operators.
+-- So you need to be cautious when opening this guc.
+set optimizer_agg_pds_strategy to 1;
+explain (costs off) select v1,v2,v3 from pds_t1 group by v3,v2,v1;
+set optimizer_agg_pds_strategy to 0;
+explain (costs off) select v1,v2,v3 from pds_t1 group by v3,v2,v1;
+
+reset optimizer_agg_pds_strategy;