Patch for Delta Materialized View Refreshes

diff --git a/src/backend/commands/matview.c b/src/backend/commands/matview.c
index e1eb7c3..da525d4 100644
--- a/src/backend/commands/matview.c
+++ b/src/backend/commands/matview.c
@@ -31,6 +31,9 @@
 #include "executor/executor.h"
 #include "executor/spi.h"
 #include "miscadmin.h"
+#include "nodes/makefuncs.h"
+#include "optimizer/clauses.h"
+#include "parser/parse_clause.h"
 #include "parser/parse_relation.h"
 #include "pgstat.h"
 #include "rewrite/rewriteHandler.h"
@@ -40,6 +43,7 @@
 #include "utils/builtins.h"
 #include "utils/lsyscache.h"
 #include "utils/rel.h"
+#include "utils/ruleutils.h"
 #include "utils/snapmgr.h"
 #include "utils/syscache.h"
 
@@ -62,10 +66,10 @@ static bool transientrel_receive(TupleTableSlot *slot, DestReceiver *self);
 static void transientrel_shutdown(DestReceiver *self);
 static void transientrel_destroy(DestReceiver *self);
 static uint64 refresh_matview_datafill(DestReceiver *dest, Query *query,
-						 const char *queryString);
+						 const char *queryString, Node *filterClause);
 static char *make_temptable_name_n(char *tempname, int n);
 static void refresh_by_match_merge(Oid matviewOid, Oid tempOid, Oid relowner,
-					   int save_sec_context);
+					   int save_sec_context, const char *filterClauseStr);
 static void refresh_by_heap_swap(Oid matviewOid, Oid OIDNewHeap, char relpersistence);
 static bool is_usable_unique_index(Relation indexRel);
 static void OpenMatViewIncrementalMaintenance(void);
@@ -111,6 +115,61 @@ SetMatViewPopulatedState(Relation relation, bool newstate)
 	CommandCounterIncrement();
 }
 
+/*
+ * Given a subquery, and a filter clause, create the equivalent of
+ * SELECT * FROM <subquery> WHERE <filter clause>
+ */
+static Query *
+mv_refresh_build_subquery_with_filter (Query *subquery,
+								Node *filterClause)
+{
+	Query *scan_query = makeNode (Query);
+	
+	scan_query->commandType = 1;
+	scan_query->querySource = QSRC_ORIGINAL;
+	scan_query->resultRelation = 0;
+	scan_query->hasAggs = false;
+	scan_query->hasWindowFuncs = false;
+	scan_query->hasTargetSRFs = false;
+	scan_query->hasSubLinks = false;
+	scan_query->hasDistinctOn = false;
+	scan_query->hasRecursive = false;
+	scan_query->hasModifyingCTE = false;
+	scan_query->hasForUpdate = false;
+	scan_query->hasRowSecurity = false; // FIXME: is this correct, or should we inherit?
+	
+	scan_query->cteList = NIL;
+	
+	/* Create a dummy ParseState for addRangeTableEntryForSubquery */
+	ParseState *pstate = make_parsestate(NULL);
+	
+	RangeTblEntry *rte = addRangeTableEntryForSubquery (pstate, subquery, makeAlias ("tab", NIL), false, false);
+	scan_query->rtable = list_make1 (rte);
+	
+	RangeTblRef *rtr = makeNode (RangeTblRef);
+	rtr->rtindex = 1;
+	
+	scan_query->jointree = makeNode (FromExpr);
+	scan_query->jointree->fromlist = list_make1 (rtr);
+	
+	scan_query->jointree->quals = filterClause;
+	
+	scan_query->targetList = expandRelAttrs (pstate, rte, 1, 0, -1);
+	
+	scan_query->override = OVERRIDING_NOT_SET;
+	scan_query->returningList = NIL;
+	scan_query->rowMarks = NIL;
+	scan_query->setOperations = NULL;
+	scan_query->constraintDeps = NIL;
+	scan_query->withCheckOptions = NIL;
+	
+	scan_query->stmt_location = -1;
+	scan_query->stmt_len = -1;
+	
+	return scan_query;
+}
+
+
 /*
  * ExecRefreshMatView -- execute a REFRESH MATERIALIZED VIEW command
  *
@@ -247,6 +306,43 @@ ExecRefreshMatView(RefreshMatViewStmt *stmt, const char *queryString,
 					 errhint("Create a unique index with no WHERE clause on one or more columns of the materialized view.")));
 	}
 
+	/* Construct a filter clause, if one was supplied. */
+	char *filterClauseStr = NULL;
+	Node *filterClause = NULL;
+	if (stmt->whereClause != NULL)
+	{
+		if (!concurrent)
+			ereport(ERROR,
+					(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+					 errmsg("cannot partially refresh materialized view \"%s\" that is not refreshed CONCURRENTLY",
+							quote_qualified_identifier(get_namespace_name(RelationGetNamespace(matviewRel)),
+													   RelationGetRelationName(matviewRel))),
+					 errhint("refresh the materialized view CONCURRENTLY.")));
+		
+		ParseState *pstate = make_parsestate(NULL);
+
+		RangeTblEntry *rte = addRangeTableEntryForRelation(pstate, matviewRel,
+														   NULL, false, false);
+		addRTEtoQuery(pstate, rte, false, true, true);
+		
+		filterClause = transformWhereClause(pstate,
+										  copyObject(stmt->whereClause),
+										  EXPR_KIND_WHERE,
+										  "WHERE");
+
+		if (contain_volatile_functions(filterClause))
+			ereport(ERROR,
+					(errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
+					 errmsg("cannot partially refresh materialized view \"%s\" with volatile expresions in the WHERE clause",
+							quote_qualified_identifier(get_namespace_name(RelationGetNamespace(matviewRel)),
+													   RelationGetRelationName(matviewRel))),
+					 errhint("refresh the materialized view without a WHERE clause, or adjust the expressions or function definitions.")));
+
+		filterClauseStr = deparse_expression (filterClause,
+												deparse_context_for ("tab", matviewOid),
+												true, false);
+	}
+	
 	/*
 	 * The stored query was rewritten at the time of the MV definition, but
 	 * has not been scribbled on by the planner.
@@ -311,7 +407,7 @@ ExecRefreshMatView(RefreshMatViewStmt *stmt, const char *queryString,
 
 	/* Generate the data, if wanted. */
 	if (!stmt->skipData)
-		processed = refresh_matview_datafill(dest, dataQuery, queryString);
+		processed = refresh_matview_datafill(dest, dataQuery, queryString, filterClause);
 
 	/* Make the matview match the newly generated data. */
 	if (concurrent)
@@ -321,7 +417,7 @@ ExecRefreshMatView(RefreshMatViewStmt *stmt, const char *queryString,
 		PG_TRY();
 		{
 			refresh_by_match_merge(matviewOid, OIDNewHeap, relowner,
-								   save_sec_context);
+								   save_sec_context, filterClauseStr);
 		}
 		PG_CATCH();
 		{
@@ -365,11 +461,15 @@ ExecRefreshMatView(RefreshMatViewStmt *stmt, const char *queryString,
  * Execute the given query, sending result rows to "dest" (which will
  * insert them into the target matview).
  *
+ * If filterCluase is provided, the query is executed as a subsquery,
+ * and the filterClause is applied as a WHERE clause at the top
+ * level.
+ *
  * Returns number of rows inserted.
  */
 static uint64
 refresh_matview_datafill(DestReceiver *dest, Query *query,
-						 const char *queryString)
+						 const char *queryString, Node *filterClause)
 {
 	List	   *rewritten;
 	PlannedStmt *plan;
@@ -389,6 +489,12 @@ refresh_matview_datafill(DestReceiver *dest, Query *query,
 
 	/* Check for user-requested abort. */
 	CHECK_FOR_INTERRUPTS();
+	
+	/* Construct as subquery and apply filter, if filter is provided. */
+	if (filterClause != NULL)
+	{
+		query = mv_refresh_build_subquery_with_filter (query, filterClause);
+	}
 
 	/* Plan the query which will generate data for the refresh. */
 	plan = pg_plan_query(query, 0, NULL);
@@ -575,10 +681,13 @@ make_temptable_name_n(char *tempname, int n)
  * incremental maintenance.  It also doesn't seem reasonable or safe to allow
  * SELECT FOR UPDATE or SELECT FOR SHARE on rows being updated or deleted by
  * this command.
+ *
+ * If filterCluaseStr[ing] is provided, the join is executed aginst the contents
+ * of the MV, filtered by the supplied clause.
  */
 static void
 refresh_by_match_merge(Oid matviewOid, Oid tempOid, Oid relowner,
-					   int save_sec_context)
+					   int save_sec_context, const char *filterClauseStr)
 {
 	StringInfoData querybuf;
 	Relation	matviewRel;
@@ -656,8 +765,38 @@ refresh_by_match_merge(Oid matviewOid, Oid tempOid, Oid relowner,
 	appendStringInfo(&querybuf,
 					 "CREATE TEMP TABLE %s AS "
 					 "SELECT mv.ctid AS tid, newdata "
-					 "FROM %s mv FULL JOIN %s newdata ON (",
-					 diffname, matviewname, tempname);
+					 "FROM ("
+					 "  SELECT tab.ctid, row (tab.*) tab_row",
+					 diffname);
+
+	/*
+	 * Append each of the columns of the table, but name them "attrN" in order
+	 * that the name does not conflict with either ctid (which won't happen) or
+	 * tab_row (which might).
+	 */
+	 tupdesc = RelationGetDescr (matviewRel);
+	 for (int attrnum = 1; attrnum <= RelationGetNumberOfAttributes (matviewRel); attrnum++)
+	 {
+		 Form_pg_attribute attr = TupleDescAttr (tupdesc, attrnum - 1);
+		 appendStringInfo(&querybuf,
+						  ", %s AS attr%d",
+						  quote_qualified_identifier("tab", NameStr(attr->attname)),
+						  attrnum);
+	 }
+
+	appendStringInfo(&querybuf,
+					 " FROM %s tab",
+					 matviewname);
+
+	/* Append the filter string, if one was supplied. */
+	if (filterClauseStr != NULL)
+	{
+		appendStringInfo(&querybuf, " WHERE %s", filterClauseStr);
+	}
+	
+	appendStringInfo(&querybuf,
+					 ") mv FULL JOIN %s newdata ON (",
+					 tempname);
 
 	/*
 	 * Get the list of index OIDs for the table from the relcache, and look up
@@ -665,7 +804,6 @@ refresh_by_match_merge(Oid matviewOid, Oid tempOid, Oid relowner,
 	 * columns present in all unique indexes which only reference columns and
 	 * include all rows.
 	 */
-	tupdesc = matviewRel->rd_att;
 	opUsedForQual = (Oid *) palloc0(sizeof(Oid) * relnatts);
 	foundUniqueIndex = false;
 
@@ -751,8 +889,8 @@ refresh_by_match_merge(Oid matviewOid, Oid tempOid, Oid relowner,
 
 				leftop = quote_qualified_identifier("newdata",
 													NameStr(attr->attname));
-				rightop = quote_qualified_identifier("mv",
-													 NameStr(attr->attname));
+				/* Name the right operand according to its attribute number. */
+				rightop = psprintf ("mv.attr%d", attnum);
 
 				generate_operator_clause(&querybuf,
 										 leftop, attrtype,
@@ -779,7 +917,7 @@ refresh_by_match_merge(Oid matviewOid, Oid tempOid, Oid relowner,
 	Assert(foundUniqueIndex);
 
 	appendStringInfoString(&querybuf,
-						   " AND newdata OPERATOR(pg_catalog.*=) mv) "
+						   " AND newdata OPERATOR(pg_catalog.*=) mv.tab_row) "
 						   "WHERE newdata IS NULL OR mv IS NULL "
 						   "ORDER BY tid");
 
diff --git a/src/backend/nodes/copyfuncs.c b/src/backend/nodes/copyfuncs.c
index 648758d..098c6b3 100644
--- a/src/backend/nodes/copyfuncs.c
+++ b/src/backend/nodes/copyfuncs.c
@@ -3900,6 +3900,7 @@ _copyRefreshMatViewStmt(const RefreshMatViewStmt *from)
 	COPY_SCALAR_FIELD(concurrent);
 	COPY_SCALAR_FIELD(skipData);
 	COPY_NODE_FIELD(relation);
+	COPY_NODE_FIELD(whereClause);
 
 	return newnode;
 }
diff --git a/src/backend/nodes/equalfuncs.c b/src/backend/nodes/equalfuncs.c
index 6a971d0..c12abb6 100644
--- a/src/backend/nodes/equalfuncs.c
+++ b/src/backend/nodes/equalfuncs.c
@@ -1710,6 +1710,7 @@ _equalRefreshMatViewStmt(const RefreshMatViewStmt *a, const RefreshMatViewStmt *
 	COMPARE_SCALAR_FIELD(concurrent);
 	COMPARE_SCALAR_FIELD(skipData);
 	COMPARE_NODE_FIELD(relation);
+	COMPARE_NODE_FIELD(whereClause);
 
 	return true;
 }
diff --git a/src/backend/parser/gram.y b/src/backend/parser/gram.y
index 02b500e..e01b785 100644
--- a/src/backend/parser/gram.y
+++ b/src/backend/parser/gram.y
@@ -4135,12 +4135,13 @@ OptNoLog:	UNLOGGED					{ $$ = RELPERSISTENCE_UNLOGGED; }
  *****************************************************************************/
 
 RefreshMatViewStmt:
-			REFRESH MATERIALIZED VIEW opt_concurrently qualified_name opt_with_data
+			REFRESH MATERIALIZED VIEW opt_concurrently qualified_name opt_with_data where_clause
 				{
 					RefreshMatViewStmt *n = makeNode(RefreshMatViewStmt);
 					n->concurrent = $4;
 					n->relation = $5;
 					n->skipData = !($6);
+					n->whereClause = $7;
 					$$ = (Node *) n;
 				}
 		;
diff --git a/src/include/nodes/parsenodes.h b/src/include/nodes/parsenodes.h
index a49b013..eebeee8 100644
--- a/src/include/nodes/parsenodes.h
+++ b/src/include/nodes/parsenodes.h
@@ -3211,6 +3211,7 @@ typedef struct RefreshMatViewStmt
 	NodeTag		type;
 	bool		concurrent;		/* allow concurrent access? */
 	bool		skipData;		/* true for WITH NO DATA */
+	Node	   *whereClause;	/* WHERE qualification */
 	RangeVar   *relation;		/* relation to insert into */
 } RefreshMatViewStmt;