Randomize output segment for non-master gather motion

Via https://github.com/greenplum-db/gporca/pull/400, ORCA will optimize
DML queries by enforcing a gather on segment instead of master, whenever
possible.

Previous to this commit, ORCA always picked the first segment to gather
on while translating the DXL-GatherMotion node to GPDB motion node.

This commit uses GPDB's hash function to select the segment to gather
on, in a round-robin fashion starting with a random segment index. This
will ensure that concurrent DML queries issued via a same session, will
be gathered on different segments to distribute the workload.
Signed-off-by: NDhanashree Kashid <dkashid@pivotal.io>

config/orca.m4

@@ -40,10 +40,10 @@ AC_RUN_IFELSE([AC_LANG_PROGRAM([[
 #include <string.h>
 ]],
 [
-return strncmp("2.75.", GPORCA_VERSION_STRING, 5);
+return strncmp("3.1.", GPORCA_VERSION_STRING, 4);
 ])],
 [AC_MSG_RESULT([[ok]])],
-[AC_MSG_ERROR([Your ORCA version is expected to be 2.75.XXX])]
+[AC_MSG_ERROR([Your ORCA version is expected to be 3.1.XXX])]
 )
 AC_LANG_POP([C++])
 ])# PGAC_CHECK_ORCA_VERSION

configure

@@ -12523,7 +12523,7 @@ int
 main ()
 {
 
-return strncmp("2.75.", GPORCA_VERSION_STRING, 5);
+return strncmp("3.1.", GPORCA_VERSION_STRING, 4);
 
   ;
   return 0;
@@ -12533,7 +12533,7 @@ if ac_fn_cxx_try_run "$LINENO"; then :
   { $as_echo "$as_me:${as_lineno-$LINENO}: result: ok" >&5
 $as_echo "ok" >&6; }
 else
-  as_fn_error $? "Your ORCA version is expected to be 2.75.XXX" "$LINENO" 5
+  as_fn_error $? "Your ORCA version is expected to be 3.1.XXX" "$LINENO" 5
 
 fi
 rm -f core *.core core.conftest.* gmon.out bb.out conftest$ac_exeext \

depends/conanfile_orca.txt

 [requires]
-orca/v2.75.0@gpdb/stable
+orca/v3.1.0@gpdb/stable
 
 [imports]
 include, * -> build/include

gpAux/releng/releng.mk

@@ -120,7 +120,7 @@ sync_tools: opt_write_test /opt/releng/apache-ant
 	-Divyrepo.user=$(IVYREPO_USER) -Divyrepo.passwd="$(IVYREPO_PASSWD)" -quiet resolve);
 
 ifeq "$(findstring aix,$(BLD_ARCH))" ""
-	LD_LIBRARY_PATH='' wget --no-check-certificate -q -O - https://github.com/greenplum-db/gporca/releases/download/v2.75.0/bin_orca_centos5_release.tar.gz | tar zxf - -C $(BLD_TOP)/ext/$(BLD_ARCH)
+	LD_LIBRARY_PATH='' wget --no-check-certificate -q -O - https://github.com/greenplum-db/gporca/releases/download/v3.1.0/bin_orca_centos5_release.tar.gz | tar zxf - -C $(BLD_TOP)/ext/$(BLD_ARCH)
 endif
 
 clean_tools: opt_write_test

src/backend/gpopt/gpdbwrappers.cpp

@@ -2722,6 +2722,27 @@ gpdb::CdbHashConst
 	return 0;
 }
 
+// pick a segment randomly from a pool of segments using GPDB's hash function
+int32
+gpdb::CdbHashRandom
+	(
+	int num_segments
+	)
+{
+	GP_WRAP_START;
+	{
+		CdbHash    *pcdbhash = makeCdbHash(num_segments);
+
+		cdbhashinit(pcdbhash);
+
+		cdbhashnokey(pcdbhash);
+
+		return cdbhashreduce(pcdbhash);
+	}
+	GP_WRAP_END;
+	return 0;
+}
+
 // hash a list of const values with GPDB's hash function
 int32 
 gpdb::CdbHashConstList

src/backend/gpopt/translate/CTranslatorDXLToPlStmt.cpp

@@ -2510,6 +2510,21 @@ CTranslatorDXLToPlStmt::TranslateDXLMotion
 			motion->motionType = MOTIONTYPE_FIXED;
 			// get segment id
 			INT segid = CDXLPhysicalGatherMotion::Cast(motion_dxlop)->IOutputSegIdx();
+
+			// if it's a gather on a segment, pick a segment from
+			// available segments using GPDB's hash function.
+			// This function outputs a segment index in a round
+			// robin fashion using a random segment index as the
+			// starting point.
+			// This ensures that concurrent DML queries issued via
+			// a same session, use a different output segment each
+			// time a gather on segment is needed.
+			if (segid >= 0)
+			{
+				segid = gpdb::CdbHashRandom(m_num_of_segments);
+				GPOS_ASSERT(segid >= 0);
+			}
+
 			motion->numOutputSegs = 1;
 			motion->outputSegIdx = (INT *) gpdb::GPDBAlloc(sizeof(INT));
 			*(motion->outputSegIdx) = segid;

src/include/gpopt/gpdbwrappers.h

@@ -582,7 +582,10 @@ namespace gpdb {
 	
 	// hash a const value with GPDB's hash function
 	int32 CdbHashConst(Const *constant, int num_segments);
-	
+
+	// pick a random segment from a pool of segments using GPDB's hash function
+	int32 CdbHashRandom(int num_segments);
+
 	// hash a list of const values with GPDB's hash function
 	int32 CdbHashConstList(List *constants, int num_segments);
 	

src/test/regress/expected/gporca.out

@@ -10263,9 +10263,9 @@ create table t_new as select avg(a) from test1 join (select i from unnest(array[
 NOTICE:  Table doesn't have 'DISTRIBUTED BY' clause -- Using column(s) named 'avg' as the Greenplum Database data distribution key for this table.
 HINT:  The 'DISTRIBUTED BY' clause determines the distribution of data. Make sure column(s) chosen are the optimal data distribution key to minimize skew.
 select * from t_new;
-        avg         
---------------------
- 2.0000000000000000
+ avg 
+-----
+   2
 (1 row)
 
 -- start_ignore
@@ -10276,9 +10276,9 @@ create table t_new as select avg(a) from test1 join (select i from unnest(array[
 NOTICE:  Table doesn't have 'DISTRIBUTED BY' clause -- Using column(s) named 'avg' as the Greenplum Database data distribution key for this table.
 HINT:  The 'DISTRIBUTED BY' clause determines the distribution of data. Make sure column(s) chosen are the optimal data distribution key to minimize skew.
 select * from t_new;
-        avg         
---------------------
- 2.0000000000000000
+ avg 
+-----
+   2
 (1 row)
 
 -- start_ignore

src/test/regress/expected/gporca_optimizer.out

@@ -10323,9 +10323,9 @@ insert into test1 select generate_series(1,100),generate_series(1,100);
 create table t_new as select avg(a) from test1 join (select i from unnest(array[1,2,3]) i) t on (test1.a = t.i);
 NOTICE:  Table doesn't have 'DISTRIBUTED BY' clause. Creating a NULL policy entry.
 select * from t_new;
-        avg         
---------------------
- 2.0000000000000000
+ avg 
+-----
+   2
 (1 row)
 
 -- start_ignore
@@ -10335,9 +10335,9 @@ set optimizer_enable_gather_on_segment_for_DML=off;
 create table t_new as select avg(a) from test1 join (select i from unnest(array[1,2,3]) i) t on (test1.a = t.i);
 NOTICE:  Table doesn't have 'DISTRIBUTED BY' clause. Creating a NULL policy entry.
 select * from t_new;
-        avg         
---------------------
- 2.0000000000000000
+ avg 
+-----
+   2
 (1 row)
 
 -- start_ignore

src/test/regress/expected/update_gp_optimizer.out

@@ -108,8 +108,8 @@ WHERE t1.user_vie_project_code_pk = keo1.user_vie_project_code_pk;
                                              ->  Broadcast Motion 3:3  (slice7; segments: 3)  (cost=0.00..2155.00 rows=1 width=8)
                                                    ->  Hash Join  (cost=0.00..2155.00 rows=1 width=8)
                                                          Hash Cond: public.keo1.user_vie_project_code_pk::text = keo2.projects_pk::text
-                                                         ->  Redistribute Motion 1:3  (slice5; segments: 1)
-                                                               ->  Hash Join
+                                                         ->  Redistribute Motion 1:3  (slice5; segments: 1)  (cost=0.00..1724.00 rows=1 width=8)
+                                                               ->  Hash Join  (cost=0.00..1724.00 rows=1 width=8)
                                                                      Hash Cond: public.keo1.user_vie_fiscal_year_period_sk::text = (max(keo3.sky_per::text))
                                                                      ->  Gather Motion 3:1  (slice1; segments: 3)  (cost=0.00..431.00 rows=1 width=16)
                                                                            ->  Table Scan on keo1  (cost=0.00..431.00 rows=1 width=16)
@@ -167,21 +167,12 @@ EXPLAIN DELETE FROM keo5 WHERE x IN (SELECT x FROM keo5 WHERE EXISTS (SELECT x F
                ->  Hash  (cost=1324032.17..1324032.17 rows=1 width=4)
                      ->  Nested Loop EXISTS Join  (cost=0.00..1324032.17 rows=1 width=4)
                            Join Filter: true
-<<<<<<< HEAD
                            ->  Table Scan on keo5  (cost=0.00..431.00 rows=1 width=4)
                            ->  Materialize  (cost=0.00..431.00 rows=1 width=1)
                                  ->  Broadcast Motion 1:3  (slice2)  (cost=0.00..431.00 rows=3 width=1)
                                        ->  Limit  (cost=0.00..431.00 rows=1 width=1)
                                              ->  Gather Motion 3:1  (slice1; segments: 3)  (cost=0.00..431.00 rows=1 width=1)
                                                    ->  Table Scan on keo5  (cost=0.00..431.00 rows=1 width=1)
-=======
-                           ->  Table Scan on keo5
-                           ->  Materialize
-                                 ->  Broadcast Motion 1:3  (slice2; segments: 1)
-                                       ->  Limit
-                                             ->  Gather Motion 3:1  (slice1; segments: 3)
-                                                   ->  Table Scan on keo5
->>>>>>> a341621d64... Introduce optimizer_enable_gather_on_segment_for_DML GUC
                                                          Filter: x < 2
  Settings:  optimizer=on
  Optimizer status: PQO version 2.42.3