Add CTE support in IVM

Simple CTEs which does not contain aggregates or DISTINCT are now
supported similarly to simple sub-queries.

Before a view is maintained, all CTEs are converted to corresponding
subqueries to enable to treat CTEs as same as subqueries. For this
end, codes of the static fnction inline_cte in the core
(optimizer/plan/subselect.c) was imported.

Makefile

@@ -6,7 +6,8 @@ OBJS = \
 	createas.o \
 	matview.o \
 	pg_ivm.o \
-	ruleutils.o
+	ruleutils.o \
+	subselect.o
 PGFILEDESC = "pg_ivm - incremental view maintenance on PostgreSQL"
 
 EXTENSION = pg_ivm

createas.c

@@ -515,8 +515,15 @@ CreateIvmTriggersOnBaseTablesRecurse(Query *qry, Node *node, Oid matviewOid,
 		case T_Query:
 			{
 				Query *query = (Query *) node;
+				ListCell *lc;
 
 				CreateIvmTriggersOnBaseTablesRecurse(qry, (Node *)query->jointree, matviewOid, relids, ex_lock);
+				foreach(lc, query->cteList)
+				{
+					CommonTableExpr *cte = (CommonTableExpr *) lfirst(lc);
+					Assert(IsA(cte->ctequery, Query));
+					CreateIvmTriggersOnBaseTablesRecurse((Query *) cte->ctequery, cte->ctequery, matviewOid, relids, ex_lock);
+				}
 			}
 			break;
 
@@ -699,11 +706,6 @@ check_ivm_restriction_walker(Node *node, check_ivm_restriction_context *context)
 				ListCell   *lc;
 				List       *vars;
 
-				/* if contained CTE, return error */
-				if (qry->cteList != NIL)
-					ereport(ERROR,
-							(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
-							 errmsg("CTE is not supported on incrementally maintainable materialized view")));
 				if (qry->groupClause != NIL && !qry->hasAggs)
 					ereport(ERROR,
 							(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
@@ -744,6 +746,10 @@ check_ivm_restriction_walker(Node *node, check_ivm_restriction_context *context)
 					ereport(ERROR,
 							(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
 							 errmsg("FOR UPDATE/SHARE clause is not supported on incrementally maintainable materialized view")));
+				if (qry->hasRecursive)
+					ereport(ERROR,
+							(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+							 errmsg("recursive CTE is not supported on incrementally maintainable materialized view")));
 
 				/* system column restrictions */
 				vars = pull_vars_of_level((Node *) qry, 0);
@@ -828,6 +834,20 @@ check_ivm_restriction_walker(Node *node, check_ivm_restriction_context *context)
 
 				query_tree_walker(qry, check_ivm_restriction_walker, (void *) context, QTW_IGNORE_RT_SUBQUERIES);
 
+				break;
+			}
+		case T_CommonTableExpr:
+			{
+				CommonTableExpr *cte = (CommonTableExpr *) node;
+
+				if (isIvmName(cte->ctename))
+					ereport(ERROR,
+							(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+							 errmsg("CTE name %s is not supported on incrementally maintainable materialized view", cte->ctename)));
+
+				context->sublevels_up++;
+				check_ivm_restriction_walker(cte->ctequery, (void *) context);
+				context->sublevels_up--;
 				break;
 			}
 		case T_TargetEntry:
@@ -1301,6 +1321,21 @@ get_primary_key_attnos_from_query(Query *query, List **constraintList, bool is_c
 	Bitmapset *keys = NULL;
 	Relids	rels_in_from;
 
+	/* convert CTEs to subqueries */
+	query = copyObject(query);
+	foreach (lc, query->cteList)
+	{
+		PlannerInfo root;
+		CommonTableExpr *cte = (CommonTableExpr *) lfirst(lc);
+
+		if (cte->cterefcount == 0)
+			continue;
+
+		root.parse = query;
+		inline_cte(&root, cte);
+	}
+	query->cteList = NIL;
+
 	/*
 	 * Collect primary key attributes from all tables used in query. The key attributes
 	 * sets for each table are stored in key_attnos_list in order by RTE index.

expected/pg_ivm.out

@@ -739,6 +739,140 @@ SELECT * FROM mv_ivm_join_subquery ORDER BY i,j,k;
  3 | 30 | 133
 (8 rows)
 
+ROLLBACK;
+-- support simple CTE
+BEGIN;
+SELECT create_immv('mv_cte',
+    'WITH b AS ( SELECT * FROM mv_base_b) SELECT a.i,a.j FROM mv_base_a a, b WHERE a.i = b.i');
+NOTICE:  could not create an index on immv "mv_cte" automatically
+DETAIL:  This target list does not have all the primary key columns, or this view does not contain GROUP BY or DISTINCT clause.
+HINT:  Create an index on the immv for efficient incremental maintenance.
+ create_immv 
+-------------
+           4
+(1 row)
+
+INSERT INTO mv_base_a VALUES(2,20);
+INSERT INTO mv_base_b VALUES(3,300);
+SELECT * FROM mv_cte ORDER BY i,j;
+ i | j  
+---+----
+ 1 | 10
+ 2 | 20
+ 2 | 20
+ 3 | 30
+ 3 | 30
+ 4 | 40
+(6 rows)
+
+ROLLBACK;
+BEGIN;
+SELECT create_immv('mv_cte',
+    'WITH a AS (SELECT * FROM mv_base_a), b AS ( SELECT * FROM mv_base_b) SELECT a.i,a.j FROM a, b WHERE a.i = b.i');
+NOTICE:  could not create an index on immv "mv_cte" automatically
+DETAIL:  This target list does not have all the primary key columns, or this view does not contain GROUP BY or DISTINCT clause.
+HINT:  Create an index on the immv for efficient incremental maintenance.
+ create_immv 
+-------------
+           4
+(1 row)
+
+INSERT INTO mv_base_a VALUES(2,20);
+INSERT INTO mv_base_b VALUES(3,300);
+SELECT * FROM mv_cte ORDER BY i,j;
+ i | j  
+---+----
+ 1 | 10
+ 2 | 20
+ 2 | 20
+ 3 | 30
+ 3 | 30
+ 4 | 40
+(6 rows)
+
+ROLLBACK;
+BEGIN;
+SELECT create_immv('mv_cte',
+    'WITH b AS ( SELECT * FROM mv_base_b) SELECT v.i,v.j FROM (WITH a AS (SELECT * FROM mv_base_a) SELECT a.i,a.j FROM a, b WHERE a.i = b.i) v');
+NOTICE:  could not create an index on immv "mv_cte" automatically
+DETAIL:  This target list does not have all the primary key columns, or this view does not contain GROUP BY or DISTINCT clause.
+HINT:  Create an index on the immv for efficient incremental maintenance.
+ create_immv 
+-------------
+           4
+(1 row)
+
+INSERT INTO mv_base_a VALUES(2,20);
+INSERT INTO mv_base_b VALUES(3,300);
+SELECT * FROM mv_cte ORDER BY i,j;
+ i | j  
+---+----
+ 1 | 10
+ 2 | 20
+ 2 | 20
+ 3 | 30
+ 3 | 30
+ 4 | 40
+(6 rows)
+
+ROLLBACK;
+BEGIN;
+SELECT create_immv('mv_cte',
+    'SELECT * FROM (WITH a AS (SELECT * FROM mv_base_a), b AS ( SELECT * FROM mv_base_b) SELECT a.i,a.j FROM a, b WHERE a.i = b.i) v');
+NOTICE:  could not create an index on immv "mv_cte" automatically
+DETAIL:  This target list does not have all the primary key columns, or this view does not contain GROUP BY or DISTINCT clause.
+HINT:  Create an index on the immv for efficient incremental maintenance.
+ create_immv 
+-------------
+           4
+(1 row)
+
+INSERT INTO mv_base_a VALUES(2,20);
+INSERT INTO mv_base_b VALUES(3,300);
+SELECT * FROM mv_cte ORDER BY i,j;
+ i | j  
+---+----
+ 1 | 10
+ 2 | 20
+ 2 | 20
+ 3 | 30
+ 3 | 30
+ 4 | 40
+(6 rows)
+
+ROLLBACK;
+BEGIN;
+SELECT create_immv('mv_cte',
+    'WITH x AS ( SELECT i, a.j, b.k FROM mv_base_b b INNER JOIN mv_base_a a USING(i)) SELECT * FROM x');
+NOTICE:  could not create an index on immv "mv_cte" automatically
+DETAIL:  This target list does not have all the primary key columns, or this view does not contain GROUP BY or DISTINCT clause.
+HINT:  Create an index on the immv for efficient incremental maintenance.
+ create_immv 
+-------------
+           4
+(1 row)
+
+WITH
+ ai AS (INSERT INTO mv_base_a VALUES (1,11),(2,22) RETURNING 0),
+ bi AS (INSERT INTO mv_base_b VALUES (1,111),(3,133) RETURNING 0),
+ bd AS (DELETE FROM mv_base_b WHERE i = 4 RETURNING 0)
+SELECT;
+--
+(1 row)
+
+SELECT * FROM mv_cte ORDER BY i,j,k;
+ i | j  |  k  
+---+----+-----
+ 1 | 10 | 101
+ 1 | 10 | 111
+ 1 | 11 | 101
+ 1 | 11 | 111
+ 2 | 20 | 102
+ 2 | 22 | 102
+ 3 | 30 | 103
+ 3 | 30 | 133
+(8 rows)
+
 ROLLBACK;
 -- views including NULL
 BEGIN;
@@ -920,7 +1054,14 @@ ERROR:  OUTER JOIN is not supported on incrementally maintainable materialized v
 -- CTE is not supported
 SELECT create_immv('mv',
     'WITH b AS ( SELECT * FROM mv_base_b) SELECT a.i,a.j FROM mv_base_a a, b WHERE a.i = b.i;');
-ERROR:  CTE is not supported on incrementally maintainable materialized view
+NOTICE:  could not create an index on immv "mv" automatically
+DETAIL:  This target list does not have all the primary key columns, or this view does not contain GROUP BY or DISTINCT clause.
+HINT:  Create an index on the immv for efficient incremental maintenance.
+ create_immv 
+-------------
+           4
+(1 row)
+
 -- contain system column
 SELECT create_immv('mv_ivm01', 'SELECT i,j,xmin FROM mv_base_a');
 ERROR:  system column is not supported on incrementally maintainable materialized view
@@ -1162,7 +1303,17 @@ SELECT immvrelid, get_immv_def(immvrelid) FROM pg_ivm_immv ORDER BY 1;
            |     b.k                         +
            |    FROM (mv_base_a a            +
            |      JOIN mv_base_b b USING (i))
-(1 row)
+ mv        |  WITH b AS (                    +
+           |          SELECT mv_base_b.i,    +
+           |             mv_base_b.k         +
+           |            FROM mv_base_b       +
+           |         )                       +
+           |  SELECT a.i,                    +
+           |     a.j                         +
+           |    FROM mv_base_a a,            +
+           |     b                           +
+           |   WHERE (a.i = b.i)
+(2 rows)
 
 -- mv_base_b is not immv
 SELECT 'mv_base_b'::regclass, get_immv_def('mv_base_b');
@@ -1172,8 +1323,9 @@ SELECT 'mv_base_b'::regclass, get_immv_def('mv_base_b');
 (1 row)
 
 DROP TABLE mv_base_b CASCADE;
-NOTICE:  drop cascades to 3 other objects
+NOTICE:  drop cascades to 4 other objects
 DETAIL:  drop cascades to table mv_ivm_1
+drop cascades to table mv
 drop cascades to view b_view
 drop cascades to materialized view b_mview
 DROP TABLE mv_base_a CASCADE;

matview.c

@@ -1128,6 +1128,20 @@ rewrite_query_for_preupdate_state(Query *query, List *tables,
 	/* XXX: Is necessary? Is this right timing? */
 	AcquireRewriteLocks(query, true, false);
 
+	/* convert CTEs to subqueries */
+	foreach (lc, query->cteList)
+	{
+		PlannerInfo root;
+		CommonTableExpr *cte = (CommonTableExpr *) lfirst(lc);
+
+		if (cte->cterefcount == 0)
+			continue;
+
+		root.parse = query;
+		inline_cte(&root, cte);
+	}
+	query->cteList = NIL;
+
 	i = 1;
 	foreach(lc, query->rtable)
 	{

pg_ivm.h

@@ -15,6 +15,7 @@
 #include "catalog/objectaddress.h"
 #include "fmgr.h"
 #include "nodes/params.h"
+#include "nodes/pathnodes.h"
 #include "parser/parse_node.h"
 #include "tcop/dest.h"
 #include "utils/queryenvironment.h"
@@ -59,4 +60,7 @@ extern bool isIvmName(const char *s);
 
 extern char *pg_ivm_get_viewdef(Relation immvrel, bool pretty);
 
+/* subselect.c */
+extern void inline_cte(PlannerInfo *root, CommonTableExpr *cte);
+
 #endif

sql/pg_ivm.sql

@@ -257,6 +257,50 @@ SELECT * FROM mv_ivm_join_subquery ORDER BY i,j,k;
 
 ROLLBACK;
 
+-- support simple CTE
+BEGIN;
+SELECT create_immv('mv_cte',
+    'WITH b AS ( SELECT * FROM mv_base_b) SELECT a.i,a.j FROM mv_base_a a, b WHERE a.i = b.i');
+INSERT INTO mv_base_a VALUES(2,20);
+INSERT INTO mv_base_b VALUES(3,300);
+SELECT * FROM mv_cte ORDER BY i,j;
+ROLLBACK;
+
+BEGIN;
+SELECT create_immv('mv_cte',
+    'WITH a AS (SELECT * FROM mv_base_a), b AS ( SELECT * FROM mv_base_b) SELECT a.i,a.j FROM a, b WHERE a.i = b.i');
+INSERT INTO mv_base_a VALUES(2,20);
+INSERT INTO mv_base_b VALUES(3,300);
+SELECT * FROM mv_cte ORDER BY i,j;
+ROLLBACK;
+
+BEGIN;
+SELECT create_immv('mv_cte',
+    'WITH b AS ( SELECT * FROM mv_base_b) SELECT v.i,v.j FROM (WITH a AS (SELECT * FROM mv_base_a) SELECT a.i,a.j FROM a, b WHERE a.i = b.i) v');
+INSERT INTO mv_base_a VALUES(2,20);
+INSERT INTO mv_base_b VALUES(3,300);
+SELECT * FROM mv_cte ORDER BY i,j;
+ROLLBACK;
+
+BEGIN;
+SELECT create_immv('mv_cte',
+    'SELECT * FROM (WITH a AS (SELECT * FROM mv_base_a), b AS ( SELECT * FROM mv_base_b) SELECT a.i,a.j FROM a, b WHERE a.i = b.i) v');
+INSERT INTO mv_base_a VALUES(2,20);
+INSERT INTO mv_base_b VALUES(3,300);
+SELECT * FROM mv_cte ORDER BY i,j;
+ROLLBACK;
+
+BEGIN;
+SELECT create_immv('mv_cte',
+    'WITH x AS ( SELECT i, a.j, b.k FROM mv_base_b b INNER JOIN mv_base_a a USING(i)) SELECT * FROM x');
+WITH
+ ai AS (INSERT INTO mv_base_a VALUES (1,11),(2,22) RETURNING 0),
+ bi AS (INSERT INTO mv_base_b VALUES (1,111),(3,133) RETURNING 0),
+ bd AS (DELETE FROM mv_base_b WHERE i = 4 RETURNING 0)
+SELECT;
+SELECT * FROM mv_cte ORDER BY i,j,k;
+ROLLBACK;
+
 -- views including NULL
 BEGIN;
 CREATE TABLE base_t (i int, v int);

subselect.c

@@ -0,0 +1,112 @@
+/*-------------------------------------------------------------------------
+ *
+ * subselect.c
+ *	  incremental view maintenance extension
+ *	  Routines for CTE support.
+ *
+ * Portions Copyright (c) 2023, IVM Development Group
+ * Portions Copyright (c) 1996-2022, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ *-------------------------------------------------------------------------
+ */
+
+#include "postgres.h"
+
+#include "nodes/nodeFuncs.h"
+#include "rewrite/rewriteManip.h"
+
+#include "pg_ivm.h"
+
+typedef struct inline_cte_walker_context
+{
+	const char *ctename;		/* name and relative level of target CTE */
+	int			levelsup;
+	Query	   *ctequery;		/* query to substitute */
+} inline_cte_walker_context;
+
+static bool inline_cte_walker(Node *node, inline_cte_walker_context *context);
+
+/*
+ * inline_cte: convert RTE_CTE references to given CTE into RTE_SUBQUERYs
+ */
+void
+inline_cte(PlannerInfo *root, CommonTableExpr *cte)
+{
+	struct inline_cte_walker_context context;
+
+	context.ctename = cte->ctename;
+	/* Start at levelsup = -1 because we'll immediately increment it */
+	context.levelsup = -1;
+	context.ctequery = castNode(Query, cte->ctequery);
+
+	(void) inline_cte_walker((Node *) root->parse, &context);
+}
+
+static bool
+inline_cte_walker(Node *node, inline_cte_walker_context *context)
+{
+	if (node == NULL)
+		return false;
+	if (IsA(node, Query))
+	{
+		Query	   *query = (Query *) node;
+
+		context->levelsup++;
+
+		/*
+		 * Visit the query's RTE nodes after their contents; otherwise
+		 * query_tree_walker would descend into the newly inlined CTE query,
+		 * which we don't want.
+		 */
+		(void) query_tree_walker(query, inline_cte_walker, context,
+								 QTW_EXAMINE_RTES_AFTER);
+
+		context->levelsup--;
+
+		return false;
+	}
+	else if (IsA(node, RangeTblEntry))
+	{
+		RangeTblEntry *rte = (RangeTblEntry *) node;
+
+		if (rte->rtekind == RTE_CTE &&
+			strcmp(rte->ctename, context->ctename) == 0 &&
+			rte->ctelevelsup == context->levelsup)
+		{
+			/*
+			 * Found a reference to replace.  Generate a copy of the CTE query
+			 * with appropriate level adjustment for outer references (e.g.,
+			 * to other CTEs).
+			 */
+			Query	   *newquery = copyObject(context->ctequery);
+
+			if (context->levelsup > 0)
+				IncrementVarSublevelsUp((Node *) newquery, context->levelsup, 1);
+
+			/*
+			 * Convert the RTE_CTE RTE into a RTE_SUBQUERY.
+			 *
+			 * Historically, a FOR UPDATE clause has been treated as extending
+			 * into views and subqueries, but not into CTEs.  We preserve this
+			 * distinction by not trying to push rowmarks into the new
+			 * subquery.
+			 */
+			rte->rtekind = RTE_SUBQUERY;
+			rte->subquery = newquery;
+			rte->security_barrier = false;
+
+			/* Zero out CTE-specific fields */
+			rte->ctename = NULL;
+			rte->ctelevelsup = 0;
+			rte->self_reference = false;
+			rte->coltypes = NIL;
+			rte->coltypmods = NIL;
+			rte->colcollations = NIL;
+		}
+
+		return false;
+	}
+
+	return expression_tree_walker(node, inline_cte_walker, context);
+}