[PATCH][postgres_fdw] Add push down of CASE WHEN clauses

diff --git a/contrib/postgres_fdw/deparse.c b/contrib/postgres_fdw/deparse.c
index 31919fda8c..8b8ca91e25 100644
--- a/contrib/postgres_fdw/deparse.c
+++ b/contrib/postgres_fdw/deparse.c
@@ -185,6 +185,7 @@ static void appendAggOrderBy(List *orderList, List *targetList,
 static void appendFunctionName(Oid funcid, deparse_expr_cxt *context);
 static Node *deparseSortGroupClause(Index ref, List *tlist, bool force_colno,
 									deparse_expr_cxt *context);
+static void deparseCaseExpr(CaseExpr *node, deparse_expr_cxt *context);
 
 /*
  * Helper functions
@@ -796,6 +797,53 @@ foreign_expr_walker(Node *node,
 					state = FDW_COLLATE_UNSAFE;
 			}
 			break;
+		case T_CaseTestExpr:
+			{
+				CaseTestExpr *c = (CaseTestExpr *) node;
+
+				/*
+				 * If the expression has nondefault collation, either it's of a
+				 * non-builtin type, or it reflects folding of a CollateExpr.
+				 * It's unsafe to send to the remote unless it's used in a
+				 * non-collation-sensitive context.
+				 */
+				collation = c->collation;
+				if (collation == InvalidOid ||
+					collation == DEFAULT_COLLATION_OID)
+					state = FDW_COLLATE_NONE;
+				else
+					state = FDW_COLLATE_UNSAFE;
+			}
+			break;
+		case T_CaseExpr:
+			{
+				ListCell   *lc;
+
+				/* Recurse to case clause subexpressions. */
+				foreach(lc, ((CaseExpr *) node)->args)
+				{
+					if (!foreign_expr_walker((Node *) lfirst(lc),
+											 glob_cxt, &inner_cxt))
+						return false;
+				}
+			}
+			break;
+		case T_CaseWhen:
+			{
+				CaseWhen   *whenExpr = (CaseWhen *) node;
+
+				/* Recurse to case clause expression. */
+				if (!foreign_expr_walker((Node *) whenExpr->expr,
+										 glob_cxt, &inner_cxt))
+					return false;
+				/* Recurse to result expression. */
+				if (!foreign_expr_walker((Node *) whenExpr->result,
+										 glob_cxt, &inner_cxt))
+					return false;
+				/* Don't apply exprType() to the case when expr. */
+				check_type = false;
+			}
+			break;
 		default:
 
 			/*
@@ -2462,6 +2510,9 @@ deparseExpr(Expr *node, deparse_expr_cxt *context)
 		case T_Aggref:
 			deparseAggref((Aggref *) node, context);
 			break;
+		case T_CaseExpr:
+			deparseCaseExpr((CaseExpr *) node, context);
+			break;
 		default:
 			elog(ERROR, "unsupported expression type for deparse: %d",
 				 (int) nodeTag(node));
@@ -3419,6 +3470,51 @@ deparseSortGroupClause(Index ref, List *tlist, bool force_colno,
 	return (Node *) expr;
 }
 
+/*
+ * Deparse CASE expression
+ */
+static void
+deparseCaseExpr(CaseExpr *node, deparse_expr_cxt *context)
+{
+	StringInfo      buf = context->buf;
+	ListCell   *lc = NULL;
+
+	appendStringInfoString(buf, "CASE");
+
+	/* If CASE arg WHEN then appen arg before continuing */
+	if (node->arg != NULL)
+	{
+		appendStringInfoString(buf, " ");
+		deparseExpr(node->arg, context);
+	}
+
+	/* Add individual cases */
+	foreach(lc, node->args)
+	{
+		CaseWhen   *whenclause = (CaseWhen *) lfirst(lc);
+
+		/* WHEN */
+		appendStringInfoString(buf, " WHEN ");
+		if (node->arg == NULL)  /* CASE WHEN */
+			deparseExpr(whenclause->expr, context);
+		else                                    /* CASE arg WHEN */
+			deparseExpr(lsecond(((OpExpr *) whenclause->expr)->args), context);
+
+		/* THEN */
+		appendStringInfoString(buf, " THEN ");
+		deparseExpr(whenclause->result, context);
+	}
+
+	/* add ELSE if needed */
+	if (node->defresult != NULL)
+	{
+		appendStringInfoString(buf, " ELSE ");
+		deparseExpr(node->defresult, context);
+	}
+
+	/* append END */
+	appendStringInfoString(buf, " END");
+}
 
 /*
  * Returns true if given Var is deparsed as a subquery output column, in
diff --git a/contrib/postgres_fdw/expected/postgres_fdw.out b/contrib/postgres_fdw/expected/postgres_fdw.out
index 25112df916..0a11463bc6 100644
--- a/contrib/postgres_fdw/expected/postgres_fdw.out
+++ b/contrib/postgres_fdw/expected/postgres_fdw.out
@@ -6153,6 +6153,63 @@ SELECT * FROM ft1 ORDER BY c6 ASC NULLS FIRST, c1 OFFSET 15 LIMIT 10;
    40 |  42 | 00040_trig_update | Tue Feb 10 00:00:00 1970 PST | Tue Feb 10 00:00:00 1970 | 0    | 0          | foo
 (10 rows)
 
+-- CASE clause push down
+EXPLAIN (VERBOSE, COSTS OFF)
+SELECT sum(CASE WHEN mod(c1, 4) = 0 THEN 1 ELSE 2 END) FROM ft1;
+                                         QUERY PLAN                                         
+--------------------------------------------------------------------------------------------
+ Foreign Scan
+   Output: (sum(CASE WHEN (mod(c1, 4) = 0) THEN 1 ELSE 2 END))
+   Relations: Aggregate on (public.ft1)
+   Remote SQL: SELECT sum(CASE WHEN (mod("C 1", 4) = 0) THEN 1 ELSE 2 END) FROM "S 1"."T 1"
+(4 rows)
+
+EXPLAIN (VERBOSE, COSTS OFF)
+SELECT * FROM ft1 WHERE c1 > (CASE WHEN mod(c1, 4) = 0 THEN 1 ELSE 100 END);
+                                                                 QUERY PLAN                                                                  
+---------------------------------------------------------------------------------------------------------------------------------------------
+ Foreign Scan on public.ft1
+   Output: c1, c2, c3, c4, c5, c6, c7, c8
+   Remote SQL: SELECT "C 1", c2, c3, c4, c5, c6, c7, c8 FROM "S 1"."T 1" WHERE (("C 1" > CASE WHEN (mod("C 1", 4) = 0) THEN 1 ELSE 100 END))
+(3 rows)
+
+EXPLAIN (VERBOSE, COSTS OFF)
+SELECT * FROM ft1 WHERE c1 > (CASE mod(c1, 4) WHEN 0 THEN 1 WHEN 2 THEN 50 ELSE 100 END);
+                                                                       QUERY PLAN                                                                       
+--------------------------------------------------------------------------------------------------------------------------------------------------------
+ Foreign Scan on public.ft1
+   Output: c1, c2, c3, c4, c5, c6, c7, c8
+   Remote SQL: SELECT "C 1", c2, c3, c4, c5, c6, c7, c8 FROM "S 1"."T 1" WHERE (("C 1" > CASE mod("C 1", 4) WHEN 0 THEN 1 WHEN 2 THEN 50 ELSE 100 END))
+(3 rows)
+
+EXPLAIN (VERBOSE, COSTS OFF)
+SELECT * FROM ft1 WHERE c1 > (CASE WHEN mod(c1, 4) = 0 THEN 1 WHEN mod(c1, 6) = 0 THEN 50 ELSE 100 END);
+                                                                                  QUERY PLAN                                                                                  
+------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
+ Foreign Scan on public.ft1
+   Output: c1, c2, c3, c4, c5, c6, c7, c8
+   Remote SQL: SELECT "C 1", c2, c3, c4, c5, c6, c7, c8 FROM "S 1"."T 1" WHERE (("C 1" > CASE WHEN (mod("C 1", 4) = 0) THEN 1 WHEN (mod("C 1", 6) = 0) THEN 50 ELSE 100 END))
+(3 rows)
+
+EXPLAIN (VERBOSE, COSTS OFF)
+SELECT * FROM ft1 WHERE substr(c3, 6) = (CASE WHEN mod(c1, 4) = 0 THEN '_trig_update' ELSE '_update' END);
+                                                                                     QUERY PLAN                                                                                     
+------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
+ Foreign Scan on public.ft1
+   Output: c1, c2, c3, c4, c5, c6, c7, c8
+   Remote SQL: SELECT "C 1", c2, c3, c4, c5, c6, c7, c8 FROM "S 1"."T 1" WHERE ((substr(c3, 6) = CASE WHEN (mod("C 1", 4) = 0) THEN '_trig_update'::text ELSE '_update'::text END))
+(3 rows)
+
+PREPARE pre_case_select AS SELECT * FROM ft1 WHERE substr(c3, 6) = (CASE WHEN mod(c1, 4) = 0 THEN $1 ELSE $2 END);
+EXPLAIN (VERBOSE, COSTS OFF) EXECUTE pre_case_select('_trig_update', '_update');
+                                                                                     QUERY PLAN                                                                                     
+------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
+ Foreign Scan on public.ft1
+   Output: c1, c2, c3, c4, c5, c6, c7, c8
+   Remote SQL: SELECT "C 1", c2, c3, c4, c5, c6, c7, c8 FROM "S 1"."T 1" WHERE ((substr(c3, 6) = CASE WHEN (mod("C 1", 4) = 0) THEN '_trig_update'::text ELSE '_update'::text END))
+(3 rows)
+
+DEALLOCATE pre_case_select;
 -- ===================================================================
 -- test check constraints
 -- ===================================================================
diff --git a/contrib/postgres_fdw/sql/postgres_fdw.sql b/contrib/postgres_fdw/sql/postgres_fdw.sql
index 95862e38ed..1b065e3c39 100644
--- a/contrib/postgres_fdw/sql/postgres_fdw.sql
+++ b/contrib/postgres_fdw/sql/postgres_fdw.sql
@@ -1008,7 +1008,6 @@ explain (verbose, costs off)
 select c2, array_agg(c2) over (partition by c2%2 order by c2 range between current row and unbounded following) from ft1 where c2 < 10 group by c2 order by 1;
 select c2, array_agg(c2) over (partition by c2%2 order by c2 range between current row and unbounded following) from ft1 where c2 < 10 group by c2 order by 1;
 
-
 -- ===================================================================
 -- parameterized queries
 -- ===================================================================
@@ -1362,6 +1361,21 @@ SELECT * FROM ft1 ORDER BY c6 DESC NULLS FIRST, c1 OFFSET 15 LIMIT 10;
 EXPLAIN (VERBOSE, COSTS OFF) SELECT * FROM ft1 ORDER BY c6 ASC NULLS FIRST, c1 OFFSET 15 LIMIT 10;
 SELECT * FROM ft1 ORDER BY c6 ASC NULLS FIRST, c1 OFFSET 15 LIMIT 10;
 
+-- CASE clause push down
+EXPLAIN (VERBOSE, COSTS OFF)
+SELECT sum(CASE WHEN mod(c1, 4) = 0 THEN 1 ELSE 2 END) FROM ft1;
+EXPLAIN (VERBOSE, COSTS OFF)
+SELECT * FROM ft1 WHERE c1 > (CASE WHEN mod(c1, 4) = 0 THEN 1 ELSE 100 END);
+EXPLAIN (VERBOSE, COSTS OFF)
+SELECT * FROM ft1 WHERE c1 > (CASE mod(c1, 4) WHEN 0 THEN 1 WHEN 2 THEN 50 ELSE 100 END);
+EXPLAIN (VERBOSE, COSTS OFF)
+SELECT * FROM ft1 WHERE c1 > (CASE WHEN mod(c1, 4) = 0 THEN 1 WHEN mod(c1, 6) = 0 THEN 50 ELSE 100 END);
+EXPLAIN (VERBOSE, COSTS OFF)
+SELECT * FROM ft1 WHERE substr(c3, 6) = (CASE WHEN mod(c1, 4) = 0 THEN '_trig_update' ELSE '_update' END);
+PREPARE pre_case_select AS SELECT * FROM ft1 WHERE substr(c3, 6) = (CASE WHEN mod(c1, 4) = 0 THEN $1 ELSE $2 END);
+EXPLAIN (VERBOSE, COSTS OFF) EXECUTE pre_case_select('_trig_update', '_update');
+DEALLOCATE pre_case_select;
+
 -- ===================================================================
 -- test check constraints
 -- ===================================================================