Fix pushing down of quals in subqueries contains window funcs

Previously, if there was a subquery contaning window functions, pushing
down of the filters was banned. This commit fixes the issue, by pushing
downs filters which are not on the columns projected using window
functions.

Adding relevant tests.

Test case:
After porting the fix to gpdb master, in the below case the filter `b = 1` is pushed down on
```
explain select b from (select b, row_number() over (partition by b) from foo) f  where b = 1;
                                             QUERY PLAN
----------------------------------------------------------------------------------------------------
 Gather Motion 3:1  (slice2; segments: 3)  (cost=0.00..1.05 rows=1 width=4)
   ->  Subquery Scan on f  (cost=0.00..1.05 rows=1 width=4)
         ->  WindowAgg  (cost=0.00..1.04 rows=1 width=4)
               ->  Redistribute Motion 3:3  (slice1; segments: 3)  (cost=0.00..1.03 rows=1 width=4)
                     Hash Key: foo.b
                     ->  Seq Scan on foo  (cost=0.00..1.01 rows=1 width=4)
                           Filter: b = 1
 Optimizer: legacy query optimizer
```

Currently on master the plan is as below where the filter is not pushed down.
```
explain select b from (select b, row_number() over (partition by b) from foo) f  where b = 1;
                                             QUERY PLAN
----------------------------------------------------------------------------------------------------
 Gather Motion 3:1  (slice2; segments: 3)  (cost=1.04..1.07 rows=1 width=4)
   ->  Subquery Scan on f  (cost=1.04..1.07 rows=1 width=4)
         Filter: f.b = 1
         ->  WindowAgg  (cost=1.04..1.06 rows=1 width=4)
               Partition By: foo.b
               ->  Sort  (cost=1.04..1.04 rows=1 width=4)
                     Sort Key: foo.b
                     ->  Redistribute Motion 3:3  (slice1; segments: 3)  (cost=0.00..1.03 rows=1 width=4)
                           Hash Key: foo.b
                           ->  Seq Scan on foo  (cost=0.00..1.01 rows=1 width=4)
 Optimizer: legacy query optimizer

src/backend/optimizer/path/allpaths.c

@@ -80,7 +80,7 @@ static bool recurse_pushdown_safe(Node *setOp, Query *topquery,
 					  bool *differentTypes);
 static void compare_tlist_datatypes(List *tlist, List *colTypes,
 						bool *differentTypes);
-static bool qual_is_pushdown_safe(Query *subquery, Index rti, Node *qual,
+static bool qual_is_pushdown_safe(Query *subquery, RangeTblEntry *rte, Index rti, Node *qual,
 					  bool *differentTypes);
 static void subquery_push_qual(Query *subquery,
 				   RangeTblEntry *rte, Index rti, Node *qual);
@@ -1429,7 +1429,7 @@ push_down_restrict(PlannerInfo *root, RelOptInfo *rel,
 			Node	   *clause = (Node *) rinfo->clause;
 
 			if (!rinfo->pseudoconstant &&
-				qual_is_pushdown_safe(subquery, rti, clause, differentTypes))
+				qual_is_pushdown_safe(subquery, rte, rti, clause, differentTypes))
 			{
 				/* Push it down */
 				subquery_push_qual(subquery, rte, rti, clause);
@@ -1487,10 +1487,6 @@ subquery_is_pushdown_safe(Query *subquery, Query *topquery,
 	if (subquery->limitOffset != NULL || subquery->limitCount != NULL)
 		return false;
 
-	/* Check point 2 */
-	if (subquery->hasWindowFuncs)
-		return false;
-
 	/* Targetlist must not contain SRF */
 	if (expression_returns_set((Node *) subquery->targetList))
 		return false;
@@ -1593,6 +1589,65 @@ compare_tlist_datatypes(List *tlist, List *colTypes,
 		elog(ERROR, "wrong number of tlist entries");
 }
 
+/*
+ * does qual include a reference to windowref node?
+ */
+static bool
+qual_contains_winref(Query *subquery, RangeTblEntry *rte, Index rti, Node *qual)
+{
+	bool result = false;
+	if (NULL != subquery && NIL != subquery->windowClause)
+	{
+		 /*
+		  * qual needs to be resolved first to map qual columns
+		  * to the underlying set of produced columns,
+		  * e.g., if we work on a setop child
+		 */
+		Node *qualNew = ResolveNew(qual, rti, 0, rte,
+								subquery->targetList,
+								CMD_SELECT, 0, NULL);
+
+		result = contain_window_function(qualNew);
+		pfree(qualNew);
+	}
+
+	return result;
+}
+
+/*
+ * is a particular qual safe to push down under set operation?
+ * if the qual contains references to windowref node, its not 
+ * safe to push it down.
+ */
+static bool
+qual_is_pushdown_safe_set_operation(RangeTblEntry *rte, Index rti, Node *qual)
+{
+
+	Query *subquery = rte->subquery;
+	Assert(subquery);
+
+	SetOperationStmt *setop = (SetOperationStmt *)subquery->setOperations;
+	Assert(setop);
+
+	/*
+	 * for queries of the form:
+	 *   SELECT * from (SELECT max(i) over () as w from X Union Select 1 as w) as foo where w > 0
+	 * the qual (w > 0) is not push_down_safe since it uses a window ref
+	 *
+	 * we check if this is the case for either left or right setop inputs
+	 *
+	 */
+	RangeTblEntry *rteLeft = rt_fetch(((RangeTblRef *)setop->larg)->rtindex, subquery->rtable);
+	RangeTblEntry *rteRight = rt_fetch(((RangeTblRef *)setop->rarg)->rtindex, subquery->rtable);
+	if (qual_contains_winref(rteLeft->subquery, rte, rti, qual) ||
+		qual_contains_winref(rteRight->subquery, rte, rti, qual))
+	{
+		return false;
+	}
+
+	return true;
+}
+
 /*
  * qual_is_pushdown_safe - is a particular qual safe to push down?
  *
@@ -1630,7 +1685,7 @@ compare_tlist_datatypes(List *tlist, List *colTypes,
  * to multiple evaluation of a volatile function.
  */
 static bool
-qual_is_pushdown_safe(Query *subquery, Index rti, Node *qual,
+qual_is_pushdown_safe(Query *subquery, RangeTblEntry *rte, Index rti, Node *qual,
 					  bool *differentTypes)
 {
 	bool		safe = true;
@@ -1647,7 +1702,11 @@ qual_is_pushdown_safe(Query *subquery, Index rti, Node *qual,
 	 * the moment we could never see any in a qual anyhow.  (The same applies
 	 * to aggregates, which we check for in pull_var_clause below.)
 	 */
-	Assert(!contain_window_function(qual));
+	if (NULL != subquery->setOperations &&
+		!qual_is_pushdown_safe_set_operation(rte, rti, qual))
+	{
+		return false;
+	}
 
 	/*
 	 * Examine all Vars used in clause; since it's a restriction clause, all

src/test/regress/expected/olap_window_seq.out

@@ -8207,3 +8207,23 @@ FROM foo;
 (30 rows)
 
 drop table foo;
+-- test predicate push down in subqueries for quals containing windowref nodes
+-- start_ignore
+create table windowagg(a int, b int);
+NOTICE:  Table doesn't have 'DISTRIBUTED BY' clause -- Using column named 'a' 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.
+-- end_ignore
+-- predicate should be pushed down
+explain select b from (select b, row_number() over (partition by b) from windowagg ) f where b = 1;
+                                               QUERY PLAN                                               
+--------------------------------------------------------------------------------------------------------
+ Gather Motion 3:1  (slice2; segments: 3)  (cost=0.00..1179.91 rows=87 width=4)
+   ->  Subquery Scan on f  (cost=0.00..1179.91 rows=29 width=4)
+         ->  WindowAgg  (cost=0.00..1179.05 rows=29 width=4)
+               ->  Redistribute Motion 3:3  (slice1; segments: 3)  (cost=0.00..1177.97 rows=29 width=4)
+                     Hash Key: windowagg.b
+                     ->  Seq Scan on windowagg  (cost=0.00..1176.25 rows=29 width=4)
+                           Filter: b = 1
+ Optimizer: legacy query optimizer
+(8 rows)
+

src/test/regress/expected/olap_window_seq_optimizer.out

@@ -8222,3 +8222,25 @@ FROM foo;
 (31 rows)
 
 drop table foo;
+-- test predicate push down in subqueries for quals containing windowref nodes
+-- start_ignore
+create table windowagg(a int, b int);
+NOTICE:  Table doesn't have 'DISTRIBUTED BY' clause -- Using column named 'a' 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.
+-- end_ignore
+-- predicate should be pushed down
+explain select b from (select b, row_number() over (partition by b) from windowagg ) f where b = 1;
+                                              QUERY PLAN                                              
+------------------------------------------------------------------------------------------------------
+ Gather Motion 3:1  (slice2; segments: 3)  (cost=0.00..431.00 rows=1 width=4)
+   ->  WindowAgg  (cost=0.00..431.00 rows=1 width=4)
+         Partition By: b
+         ->  Sort  (cost=0.00..431.00 rows=1 width=4)
+               Sort Key: b
+               ->  Redistribute Motion 3:3  (slice1; segments: 3)  (cost=0.00..431.00 rows=1 width=4)
+                     Hash Key: b
+                     ->  Table Scan on windowagg  (cost=0.00..431.00 rows=1 width=4)
+                           Filter: b = 1
+ Optimizer: PQO version 2.55.15
+(10 rows)
+

src/test/regress/sql/olap_window_seq.sql

@@ -1660,3 +1660,11 @@ EXPLAIN SELECT count(*) over (PARTITION BY a ORDER BY b, c, d) as count1,
        count(*) over (PARTITION BY a ORDER BY c, b, d) as count3
 FROM foo;
 drop table foo;
+
+-- test predicate push down in subqueries for quals containing windowref nodes
+-- start_ignore
+create table windowagg(a int, b int);
+-- end_ignore
+
+-- predicate should be pushed down
+explain select b from (select b, row_number() over (partition by b) from windowagg ) f where b = 1;