Remove unnecessary projections from duplicate sensitive Distribute(s) in ORCA

Duplicate sensitive HashDistribute Motions generated by ORCA get
translated to Result nodes with hashFilter cols set. However, if the
Motion needs to distribute based on a complex expression (rather than
just a Var), the expression must be added into the targetlist of the
Result node and then referenced in hashFilterColIdx.

However, this can affect other operators above the Result node. For
example, a Hash operator expects the targetlist of its child node to
contain only elements that are to be hashed. Additional expressions here
can cause issues with memtuple bindings that can lead to errors.

(E.g The attached test case, when run without our fix, will give an
error: "invalid input syntax for integer:")

This PR fixes the issue by adding an additional Result node on top of
the duplicate sensitive Result node to project only the elements from
the original targetlist in such cases.

src/backend/gpopt/translate/CTranslatorDXLToPlStmt.cpp

@@ -2675,6 +2675,8 @@ CTranslatorDXLToPlStmt::TranslateDXLRedistributeMotionToResultHashFilters
 		output_context
 		);
 
+	bool targetlist_modified = false;
+
 	// translate hash expr list
 	result->hashFilter = true;
 
@@ -2725,6 +2727,7 @@ CTranslatorDXLToPlStmt::TranslateDXLRedistributeMotionToResultHashFilters
 				plan->targetlist = gpdb::LAppend(plan->targetlist, target_entry);
 
 				resno = target_entry->resno;
+				targetlist_modified = true;
 			}
 			GPOS_ASSERT(gpos::int_max != resno);
 			
@@ -2740,6 +2743,66 @@ CTranslatorDXLToPlStmt::TranslateDXLRedistributeMotionToResultHashFilters
 
 	SetParamIds(plan);
 
+	Plan *child_result = (Plan *) result;
+
+	if (targetlist_modified)
+	{
+		// If the targetlist is modified by adding any expressions, such as for
+		// hashFilterColIdx & hashFilterFuncs, add an additional Result node on top
+		// to project only the elements from the original targetlist.
+		// This is needed in case the Result node is created under the Hash
+		// operator (or any non-projecting node), which expects the targetlist of its
+		// child node to contain only elements that are to be hashed.
+		// We should not generate a plan where the target list of a non-projecting
+		// node such as Hash does not match its child. Additional expressions
+		// here can cause issues with memtuple bindings that can lead to errors.
+		Result *result = MakeNode(Result);
+
+		Plan *plan = &(result->plan);
+		plan->plan_node_id = m_dxl_to_plstmt_context->GetNextPlanId();
+
+		// keep the same costs & rows estimates
+		plan->startup_cost = child_result->startup_cost;
+		plan->total_cost = child_result->total_cost;
+		plan->plan_rows = child_result->plan_rows;
+		plan->plan_width = child_result->plan_width;
+
+		// populate the targetlist based on child_result's original targetlist
+		plan->targetlist = NIL;
+		ListCell *lc = NULL;
+		ULONG ul = 0;
+		ForEach (lc, child_result->targetlist)
+		{
+			if (ul++ >= project_list_dxlnode->Arity())
+			{
+				// done with the original targetlist, stop
+				// all expressions added after project_list_dxlnode->Arity() are
+				// not output cols, but rather hash expressions and should not be projected
+				break;
+			}
+
+			TargetEntry *te = (TargetEntry *) lfirst(lc);
+			Var *var = gpdb::MakeVar(OUTER,
+									 te->resno,
+									 gpdb::ExprType((Node *) te->expr),
+									 gpdb::ExprTypeMod((Node *) te->expr),
+									 0	/* varlevelsup */);
+			TargetEntry *new_te = gpdb::MakeTargetEntry((Expr *) var,
+														ul, /* resno */
+														te->resname,
+														te->resjunk);
+			plan->targetlist = gpdb::LAppend(plan->targetlist, new_te);
+		}
+
+		plan->qual = NIL;
+		plan->lefttree = child_result;
+		plan->nMotionNodes = child_plan->nMotionNodes;
+
+		SetParamIds(plan);
+
+		return (Plan *) result;
+	}
+
 	return (Plan *) result;
 }
 

src/test/regress/expected/gporca.out

@@ -11626,3 +11626,48 @@ select * from gpexp_hash order by a;
  15 | 15
 (15 rows)
 
+drop table if exists t55;
+NOTICE:  table "t55" does not exist, skipping
+drop table if exists tp;
+NOTICE:  table "tp" does not exist, skipping
+create table t55 (c int, lid int);
+NOTICE:  Table doesn't have 'DISTRIBUTED BY' clause -- Using column named 'c' 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.
+insert into t55 select i, i from generate_series(1, 1000) i;
+set optimizer_join_order = query;
+explain
+CREATE TABLE TP AS
+WITH META AS (SELECT '2020-01-01' AS VALID_DT, '99' AS LOAD_ID)
+SELECT DISTINCT L1.c, L1.lid
+FROM t55 L1 CROSS JOIN META
+WHERE L1.lid = META.LOAD_ID;
+NOTICE:  Table doesn't have 'DISTRIBUTED BY' clause -- Using column(s) named 'c' 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.
+                                                       QUERY PLAN                                                       
+------------------------------------------------------------------------------------------------------------------------
+ HashAggregate  (cost=15.69..15.78 rows=3 width=8)
+   Group By: l1.c, l1.lid
+   ->  Hash Join  (cost=0.03..15.64 rows=4 width=8)
+         Hash Cond: l1.lid = meta.load_id::integer
+         ->  Seq Scan on t55 l1  (cost=0.00..13.00 rows=334 width=8)
+         ->  Hash  (cost=0.02..0.02 rows=1 width=32)
+               ->  Subquery Scan meta  (cost=0.00..0.02 rows=1 width=32)
+                     ->  Result  (cost=0.00..0.01 rows=1 width=0)
+ Settings:  optimizer=off; optimizer_cte_inlining_bound=1000; optimizer_join_order=query; optimizer_metadata_caching=on
+ Optimizer status: legacy query optimizer
+(10 rows)
+
+CREATE TABLE TP AS
+WITH META AS (SELECT '2020-01-01' AS VALID_DT, '99' AS LOAD_ID)
+SELECT DISTINCT L1.c, L1.lid
+FROM t55 L1 CROSS JOIN META
+WHERE L1.lid = META.LOAD_ID;
+NOTICE:  Table doesn't have 'DISTRIBUTED BY' clause -- Using column(s) named 'c' 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.
+reset optimizer_join_order;
+SELECT * from tp;
+ c  | lid 
+----+-----
+ 99 |  99
+(1 row)
+

src/test/regress/expected/gporca_optimizer.out

@@ -11611,3 +11611,54 @@ select * from gpexp_hash order by a;
  15 | 15
 (15 rows)
 
+drop table if exists t55;
+NOTICE:  table "t55" does not exist, skipping
+drop table if exists tp;
+NOTICE:  table "tp" does not exist, skipping
+create table t55 (c int, lid int);
+NOTICE:  Table doesn't have 'DISTRIBUTED BY' clause -- Using column named 'c' 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.
+insert into t55 select i, i from generate_series(1, 1000) i;
+set optimizer_join_order = query;
+explain
+CREATE TABLE TP AS
+WITH META AS (SELECT '2020-01-01' AS VALID_DT, '99' AS LOAD_ID)
+SELECT DISTINCT L1.c, L1.lid
+FROM t55 L1 CROSS JOIN META
+WHERE L1.lid = META.LOAD_ID;
+NOTICE:  Table doesn't have 'DISTRIBUTED BY' clause. Creating a NULL policy entry.
+                                                      QUERY PLAN                                                       
+-----------------------------------------------------------------------------------------------------------------------
+ Result  (cost=0.00..0.00 rows=0 width=0)
+   ->  Result  (cost=0.00..437.37 rows=134 width=8)
+         ->  Redistribute Motion 3:3  (slice2; segments: 3)  (cost=0.00..431.12 rows=134 width=8)
+               ->  HashAggregate  (cost=0.00..431.12 rows=134 width=8)
+                     Group By: c, lid
+                     ->  Hash Join  (cost=0.00..431.08 rows=134 width=8)
+                           Hash Cond: lid = pg_catalog.int4in(unknownout(load_id), 23::oid, (-1))
+                           ->  Redistribute Motion 3:3  (slice1; segments: 3)  (cost=0.00..431.02 rows=334 width=8)
+                                 Hash Key: lid
+                                 ->  Table Scan on t55  (cost=0.00..431.01 rows=334 width=8)
+                           ->  Hash  (cost=0.00..0.00 rows=1 width=8)
+                                 ->  Result  (cost=0.00..0.00 rows=1 width=8)
+                                       ->  Result  (cost=0.00..0.00 rows=1 width=8)
+                                             ->  Result  (cost=0.00..0.00 rows=1 width=8)
+                                                   ->  Result  (cost=0.00..0.00 rows=1 width=8)
+                                                         ->  Result  (cost=0.00..0.00 rows=1 width=1)
+ Settings:  optimizer=on; optimizer_cte_inlining_bound=1000; optimizer_join_order=query; optimizer_metadata_caching=on
+ Optimizer status: PQO version 3.102.0
+(18 rows)
+
+CREATE TABLE TP AS
+WITH META AS (SELECT '2020-01-01' AS VALID_DT, '99' AS LOAD_ID)
+SELECT DISTINCT L1.c, L1.lid
+FROM t55 L1 CROSS JOIN META
+WHERE L1.lid = META.LOAD_ID;
+NOTICE:  Table doesn't have 'DISTRIBUTED BY' clause. Creating a NULL policy entry.
+reset optimizer_join_order;
+SELECT * from tp;
+ c  | lid 
+----+-----
+ 99 |  99
+(1 row)
+

src/test/regress/sql/gporca.sql

@@ -1915,6 +1915,30 @@ explain update gpexp_hash o set b=(select a from gpexp_hash i where o.a = i.a) w
 update gpexp_hash o set b=(select a from gpexp_hash i where o.a = i.a) where a > 5;
 select * from gpexp_hash order by a;
 
+drop table if exists t55;
+drop table if exists tp;
+
+create table t55 (c int, lid int);
+insert into t55 select i, i from generate_series(1, 1000) i;
+
+set optimizer_join_order = query;
+
+explain
+CREATE TABLE TP AS
+WITH META AS (SELECT '2020-01-01' AS VALID_DT, '99' AS LOAD_ID)
+SELECT DISTINCT L1.c, L1.lid
+FROM t55 L1 CROSS JOIN META
+WHERE L1.lid = META.LOAD_ID;
+
+CREATE TABLE TP AS
+WITH META AS (SELECT '2020-01-01' AS VALID_DT, '99' AS LOAD_ID)
+SELECT DISTINCT L1.c, L1.lid
+FROM t55 L1 CROSS JOIN META
+WHERE L1.lid = META.LOAD_ID;
+
+reset optimizer_join_order;
+SELECT * from tp;
+
 -- start_ignore
 DROP SCHEMA orca CASCADE;
 -- end_ignore