Hi,
Attached is the latest patch for supporting self-join views. This also
including the following fix mentioned by Tatsuo Ishii.
> > On 2019-Aug-06, Tatsuo Ishii wrote:
> >
> >> It's not mentioned below but some bugs including seg fault when
> >> --enable-casser is enabled was also fixed in this patch.
> >>
> >> BTW, I found a bug with min/max support in this patch and I believe
> >> Yugo is working on it. Details:
> >> https://github.com/sraoss/pgsql-ivm/issues/20
This patch allows to support self-join views, simultaneous updates of more
than one base tables, and also multiple updates of the same base table.
I first tried to support just self-join, but I found that this is essentially
same as to support simultaneous table updates, so I decided to support them in
the same commit. I think this will be a base for implementing
Deferred-maintenance in future.
In the new implementation, AFTER triggers are used to collecting tuplestores
containing transition table contents. When multiple tables are changed,
multiple AFTER triggers are invoked, then the final AFTER trigger performs
actual update of the matview. In addition AFTER trigger, also BEFORE trigger
is used to handle global information for view maintenance.
For example, suppose that we have a view V joining table R,S, and new tuples are
inserted to each table, dR,dS, and dT respectively.
V = R*S*T
R_new = R + dR
S_new = S + dS
T_new = T + dT
In this situation, we can calculate the new view state as bellow.
V_new
= R_new * S_new * T_new
= (R + dR) * (S + dS) * (T + dT)
= R*S*T + dR*(S + dS)*(T + dT) + R*dS*(T + dT) + R*S*dT
= V + dR*(S + dS)*(T + dT) + R*dS*(T + dT) + R*S*dT
= V + (dR *S_new*T_new) + (R*dS*T_new) + (R*S*dT)
To calculate view deltas, we need both pre-state (R,S, and T) and post-state
(R_new, S_new, and T_new) of base tables.
Post-update states are available in AFTER trigger, and we calculate pre-update
states by filtering inserted tuples using cmin/xmin system columns, and
appendding
deleted tuples which are contained in a old transition table.
In the original core implementation, tuplestores of transition tables were
freed for each query depth. However, we want to prolong their life plan because
we have to preserve these for a whole query assuming some base tables are
changed
in other trigger functions, so I added a hack to trigger.c.
Regression tests are also added for self join view, multiple change on the same
table, simultaneous two table changes, and foreign reference constrains.
Here are behavior examples:
1. Table definition
- t: for self-join
- r,s: for 2-ways join
CREATE TABLE r (i int, v int);
CREATE TABLE
CREATE TABLE s (i int, v int);
CREATE TABLE
CREATE TABLE t (i int, v int);
CREATE TABLE
2. Initial data
INSERT INTO r VALUES (1, 10), (2, 20), (3, 30);
INSERT 0 3
INSERT INTO s VALUES (1, 100), (2, 200), (3, 300);
INSERT 0 3
INSERT INTO t VALUES (1, 10), (2, 20), (3, 30);
INSERT 0 3
3. View definition
3.1. self-join（mv_self, v_slef)
CREATE INCREMENTAL MATERIALIZED VIEW mv_self(v1, v2) AS
SELECT t1.v, t2.v FROM t t1 JOIN t t2 ON t1.i = t2.i;
SELECT 3
CREATE VIEW v_self(v1, v2) AS
SELECT t1.v, t2.v FROM t t1 JOIN t t2 ON t1.i = t2.i;
CREATE VIEW
3.2. 2-ways join (mv, v)
CREATE INCREMENTAL MATERIALIZED VIEW mv(v1, v2) AS
SELECT r.v, s.v FROM r JOIN s USING(i);
SELECT 3
CREATE VIEW v(v1, v2) AS
SELECT r.v, s.v FROM r JOIN s USING(i);
CREATE VIEW
3.3 Initial contents
SELECT * FROM mv_self ORDER BY v1;
v1 | v2
----+----
10 | 10
20 | 20
30 | 30
(3 rows)
SELECT * FROM mv ORDER BY v1;
v1 | v2
----+-----
10 | 100
20 | 200
30 | 300
(3 rows)
4. Update a base table for the self-join view
INSERT INTO t VALUES (4,40);
INSERT 0 1
DELETE FROM t WHERE i = 1;
DELETE 1
UPDATE t SET v = v*10 WHERE i=2;
UPDATE 1
4.1. Results
- Comparison with the normal view
SELECT * FROM mv_self ORDER BY v1;
v1 | v2
-----+-----
30 | 30
40 | 40
200 | 200
(3 rows)
SELECT * FROM v_self ORDER BY v1;
v1 | v2
-----+-----
30 | 30
40 | 40
200 | 200
(3 rows)
5. pdate a base table for the 2-way join view
WITH
ins_r AS (INSERT INTO r VALUES (1,11) RETURNING 1),
ins_r2 AS (INSERT INTO r VALUES (3,33) RETURNING 1),
ins_s AS (INSERT INTO s VALUES (2,222) RETURNING 1),
upd_r AS (UPDATE r SET v = v + 1000 WHERE i = 2 RETURNING 1),
dlt_s AS (DELETE FROM s WHERE i = 3 RETURNING 1)
SELECT NULL;
?column?
----------
(1 row)
5.1. Results
- Comparison with the normal view
SELECT * FROM mv ORDER BY v1;
v1 | v2
------+-----
10 | 100
11 | 100
1020 | 200
1020 | 222
(4 rows)
SELECT * FROM v ORDER BY v1;
v1 | v2
------+-----
10 | 100
11 | 100
1020 | 200
1020 | 222
(4 rows)
========
Best Regards,
Yugo Nagata
--
Yugo Nagata <nag...@sraoss.co.jp>
diff --git a/doc/src/sgml/catalogs.sgml b/doc/src/sgml/catalogs.sgml
index 5e71a2e865..a288bddcda 100644
--- a/doc/src/sgml/catalogs.sgml
+++ b/doc/src/sgml/catalogs.sgml
@@ -1952,6 +1952,13 @@ SCRAM-SHA-256$<replaceable><iteration count></replaceable>:<replaceable>&l
<entry>True if table or index is a partition</entry>
</row>
+ <row>
+ <entry><structfield>relisivm</structfield></entry>
+ <entry><type>bool</type></entry>
+ <entry></entry>
+ <entry>True if materialized view enables incremental view maintenance</entry>
+ </row>
+
<row>
<entry><structfield>relrewrite</structfield></entry>
<entry><type>oid</type></entry>
diff --git a/doc/src/sgml/ref/create_materialized_view.sgml b/doc/src/sgml/ref/create_materialized_view.sgml
index ec8847ed40..a23366a342 100644
--- a/doc/src/sgml/ref/create_materialized_view.sgml
+++ b/doc/src/sgml/ref/create_materialized_view.sgml
@@ -21,7 +21,7 @@ PostgreSQL documentation
<refsynopsisdiv>
<synopsis>
-CREATE MATERIALIZED VIEW [ IF NOT EXISTS ] <replaceable>table_name</replaceable>
+CREATE [ INCREMENTAL ] MATERIALIZED VIEW [ IF NOT EXISTS ] <replaceable>table_name</replaceable>
[ (<replaceable>column_name</replaceable> [, ...] ) ]
[ USING <replaceable class="parameter">method</replaceable> ]
[ WITH ( <replaceable class="parameter">storage_parameter</replaceable> [= <replaceable class="parameter">value</replaceable>] [, ... ] ) ]
@@ -54,6 +54,16 @@ CREATE MATERIALIZED VIEW [ IF NOT EXISTS ] <replaceable>table_name</replaceable>
<title>Parameters</title>
<variablelist>
+ <varlistentry>
+ <term><literal>INCREMENTAL</literal></term>
+ <listitem>
+ <para>
+ If specified, a materialized view enables incremental view maintenance.
+ You can replace only the contents of a materialized view, which based rows are changed.
+ </para>
+ </listitem>
+ </varlistentry>
+
<varlistentry>
<term><literal>IF NOT EXISTS</literal></term>
<listitem>
diff --git a/src/backend/access/transam/xact.c b/src/backend/access/transam/xact.c
index 9162286c98..c5551f59e2 100644
--- a/src/backend/access/transam/xact.c
+++ b/src/backend/access/transam/xact.c
@@ -34,6 +34,7 @@
#include "catalog/pg_enum.h"
#include "catalog/storage.h"
#include "commands/async.h"
+#include "commands/matview.h"
#include "commands/tablecmds.h"
#include "commands/trigger.h"
#include "executor/spi.h"
@@ -2661,6 +2662,7 @@ AbortTransaction(void)
AtAbort_Notify();
AtEOXact_RelationMap(false, is_parallel_worker);
AtAbort_Twophase();
+ AtAbort_IVM();
/*
* Advertise the fact that we aborted in pg_xact (assuming that we got as
diff --git a/src/backend/catalog/heap.c b/src/backend/catalog/heap.c
index b7bcdd9d0f..cb7e8be84a 100644
--- a/src/backend/catalog/heap.c
+++ b/src/backend/catalog/heap.c
@@ -890,6 +890,7 @@ InsertPgClassTuple(Relation pg_class_desc,
values[Anum_pg_class_relrewrite - 1] = ObjectIdGetDatum(rd_rel->relrewrite);
values[Anum_pg_class_relfrozenxid - 1] = TransactionIdGetDatum(rd_rel->relfrozenxid);
values[Anum_pg_class_relminmxid - 1] = MultiXactIdGetDatum(rd_rel->relminmxid);
+ values[Anum_pg_class_relisivm - 1] = BoolGetDatum(rd_rel->relisivm);
if (relacl != (Datum) 0)
values[Anum_pg_class_relacl - 1] = relacl;
else
diff --git a/src/backend/catalog/index.c b/src/backend/catalog/index.c
index 098732cc4a..5aadbf96a8 100644
--- a/src/backend/catalog/index.c
+++ b/src/backend/catalog/index.c
@@ -955,6 +955,7 @@ index_create(Relation heapRelation,
indexRelation->rd_rel->relowner = heapRelation->rd_rel->relowner;
indexRelation->rd_rel->relam = accessMethodObjectId;
indexRelation->rd_rel->relispartition = OidIsValid(parentIndexRelid);
+ indexRelation->rd_rel->relisivm = false;
/*
* store index's pg_class entry
diff --git a/src/backend/commands/createas.c b/src/backend/commands/createas.c
index b7d220699f..6f2639130c 100644
--- a/src/backend/commands/createas.c
+++ b/src/backend/commands/createas.c
@@ -51,6 +51,19 @@
#include "utils/rls.h"
#include "utils/snapmgr.h"
+#include "catalog/pg_trigger.h"
+#include "catalog/pg_type.h"
+#include "commands/trigger.h"
+#include "parser/parser.h"
+#include "parser/parsetree.h"
+#include "parser/parse_func.h"
+#include "parser/parse_type.h"
+#include "nodes/print.h"
+#include "nodes/primnodes.h"
+#include "optimizer/optimizer.h"
+#include "commands/defrem.h"
+#include "utils/syscache.h"
+
typedef struct
{
@@ -74,6 +87,9 @@ static bool intorel_receive(TupleTableSlot *slot, DestReceiver *self);
static void intorel_shutdown(DestReceiver *self);
static void intorel_destroy(DestReceiver *self);
+static void CreateIvmTrigger(Oid relOid, Oid viewOid, char *matviewname, int16 type, int16 timing);
+static void CreateIvmTriggersOnBaseTables(Query *qry, Node *jtnode, Oid matviewOid, char* matviewname, Bitmapset **relid_map);
+static void check_ivm_restriction_walker(Node *node);
/*
* create_ctas_internal
@@ -109,6 +125,8 @@ create_ctas_internal(List *attrList, IntoClause *into)
create->oncommit = into->onCommit;
create->tablespacename = into->tableSpaceName;
create->if_not_exists = false;
+ /* Using Materialized view only */
+ create->ivm = into->ivm;
create->accessMethod = into->accessMethod;
/*
@@ -239,6 +257,7 @@ ExecCreateTableAs(CreateTableAsStmt *stmt, const char *queryString,
List *rewritten;
PlannedStmt *plan;
QueryDesc *queryDesc;
+ Query *copied_query;
if (stmt->if_not_exists)
{
@@ -256,6 +275,9 @@ ExecCreateTableAs(CreateTableAsStmt *stmt, const char *queryString,
}
}
+ if (is_matview && into->ivm)
+ check_ivm_restriction_walker((Node *) query);
+
/*
* Create the tuple receiver object and insert info it will need
*/
@@ -319,7 +341,148 @@ ExecCreateTableAs(CreateTableAsStmt *stmt, const char *queryString,
* and is executed repeatedly. (See also the same hack in EXPLAIN and
* PREPARE.)
*/
- rewritten = QueryRewrite(copyObject(query));
+
+ copied_query = copyObject(query);
+ if (is_matview && into->ivm)
+ {
+ TargetEntry *tle;
+ Node *node;
+ ParseState *pstate = make_parsestate(NULL);
+ FuncCall *fn;
+
+ pstate->p_expr_kind = EXPR_KIND_SELECT_TARGET;
+
+ /* group keys must be in targetlist */
+ if (copied_query->groupClause)
+ {
+ ListCell *lc;
+ foreach(lc, copied_query->groupClause)
+ {
+ SortGroupClause *scl = (SortGroupClause *) lfirst(lc);
+ TargetEntry *tle = get_sortgroupclause_tle(scl, copied_query->targetList);
+
+ if (tle->resjunk)
+ elog(ERROR, "GROUP BY expression must appear in select list for incremental materialized views");
+ }
+ }
+ else if (!copied_query->hasAggs)
+ copied_query->groupClause = transformDistinctClause(NULL, &copied_query->targetList, copied_query->sortClause, false);
+
+ if (copied_query->hasAggs)
+ {
+ ListCell *lc;
+ List *agg_counts = NIL;
+ AttrNumber next_resno = list_length(copied_query->targetList) + 1;
+ Const *dmy_arg = makeConst(INT4OID,
+ -1,
+ InvalidOid,
+ sizeof(int32),
+ Int32GetDatum(1),
+ false,
+ true); /* pass by value */
+
+ foreach(lc, copied_query->targetList)
+ {
+ TargetEntry *tle = (TargetEntry *) lfirst(lc);
+ TargetEntry *tle_count;
+ char *resname = (into->colNames == NIL ? tle->resname : strVal(list_nth(into->colNames, tle->resno-1)));
+
+
+ if (IsA(tle->expr, Aggref))
+ {
+ Aggref *aggref = (Aggref *) tle->expr;
+ const char *aggname = get_func_name(aggref->aggfnoid);
+
+ /* XXX: need some generalization
+ *
+ * Specifically, Using func names is not robust. We can use oids instead
+ * of names, but it would be nice to add some information to pg_aggregate.
+ */
+ if (strcmp(aggname, "sum") !=0
+ && strcmp(aggname, "count") != 0
+ && strcmp(aggname, "avg") != 0
+ && strcmp(aggname, "min") != 0
+ && strcmp(aggname, "max") != 0
+ )
+ elog(ERROR, "aggregate function %s is not supported", aggname);
+
+ /*
+ * For aggregate functions except to count, add count func with the same arg parameters.
+ * Also, add sum func for agv.
+ *
+ * XXX: If there are same expressions explicitly in the target list, we can use this instead
+ * of adding new duplicated one.
+ */
+ if (strcmp(aggname, "count") != 0)
+ {
+ fn = makeFuncCall(list_make1(makeString("count")), NIL, -1);
+
+ /* Make a Func with a dummy arg, and then override this by the original agg's args. */
+ node = ParseFuncOrColumn(pstate, fn->funcname, list_make1(dmy_arg), NULL, fn, false, -1);
+ ((Aggref *)node)->args = aggref->args;
+
+ tle_count = makeTargetEntry((Expr *) node,
+ next_resno,
+ pstrdup(makeObjectName("__ivm_count",resname, "_")),
+ false);
+ agg_counts = lappend(agg_counts, tle_count);
+ next_resno++;
+ }
+ if (strcmp(aggname, "avg") == 0)
+ {
+ List *dmy_args = NIL;
+ ListCell *lc;
+ foreach(lc, aggref->aggargtypes)
+ {
+ Oid typeid = lfirst_oid(lc);
+ Type type = typeidType(typeid);
+
+ Const *con = makeConst(typeid,
+ -1,
+ typeTypeCollation(type),
+ typeLen(type),
+ (Datum) 0,
+ true,
+ typeByVal(type));
+ dmy_args = lappend(dmy_args, con);
+ ReleaseSysCache(type);
+
+ }
+ fn = makeFuncCall(list_make1(makeString("sum")), NIL, -1);
+
+ /* Make a Func with a dummy arg, and then override this by the original agg's args. */
+ node = ParseFuncOrColumn(pstate, fn->funcname, dmy_args, NULL, fn, false, -1);
+ ((Aggref *)node)->args = aggref->args;
+
+ tle_count = makeTargetEntry((Expr *) node,
+ next_resno,
+ pstrdup(makeObjectName("__ivm_sum",resname, "_")),
+ false);
+ agg_counts = lappend(agg_counts, tle_count);
+ next_resno++;
+ }
+
+ }
+ }
+ copied_query->targetList = list_concat(copied_query->targetList, agg_counts);
+
+ }
+
+ /* Add count(*) for counting algorithm */
+ fn = makeFuncCall(list_make1(makeString("count")), NIL, -1);
+ fn->agg_star = true;
+
+ node = ParseFuncOrColumn(pstate, fn->funcname, NIL, NULL, fn, false, -1);
+
+ tle = makeTargetEntry((Expr *) node,
+ list_length(copied_query->targetList) + 1,
+ pstrdup("__ivm_count__"),
+ false);
+ copied_query->targetList = lappend(copied_query->targetList, tle);
+ copied_query->hasAggs = true;
+ }
+
+ rewritten = QueryRewrite(copied_query);
/* SELECT should never rewrite to more or less than one SELECT query */
if (list_length(rewritten) != 1)
@@ -378,11 +541,77 @@ ExecCreateTableAs(CreateTableAsStmt *stmt, const char *queryString,
/* Restore userid and security context */
SetUserIdAndSecContext(save_userid, save_sec_context);
+
+
+ if (into->ivm)
+ {
+
+ Oid matviewOid = address.objectId;
+ Relation matviewRel = table_open(matviewOid, NoLock);
+ char *matviewname = quote_qualified_identifier(get_namespace_name(RelationGetNamespace(matviewRel)),
+ RelationGetRelationName(matviewRel));
+ Bitmapset *relid_map = NULL;
+
+ copied_query = copyObject(query);
+ AcquireRewriteLocks(copied_query, true, false);
+
+ CreateIvmTriggersOnBaseTables(copied_query, (Node *)copied_query->jointree, matviewOid, matviewname, &relid_map);
+
+ table_close(matviewRel, NoLock);
+
+ bms_free(relid_map);
+ }
}
return address;
}
+static void CreateIvmTriggersOnBaseTables(Query *qry, Node *jtnode, Oid matviewOid, char* matviewname, Bitmapset **relid_map)
+{
+
+ if (jtnode == NULL)
+ return;
+ if (IsA(jtnode, RangeTblRef))
+ {
+ int rti = ((RangeTblRef *) jtnode)->rtindex;
+ RangeTblEntry *rte = rt_fetch(rti, qry->rtable);
+
+ if (rte->rtekind == RTE_RELATION)
+ {
+ if (!bms_is_member(rte->relid, *relid_map))
+ {
+ CreateIvmTrigger(rte->relid, matviewOid, matviewname, TRIGGER_TYPE_INSERT, TRIGGER_TYPE_BEFORE);
+ CreateIvmTrigger(rte->relid, matviewOid, matviewname, TRIGGER_TYPE_DELETE, TRIGGER_TYPE_BEFORE);
+ CreateIvmTrigger(rte->relid, matviewOid, matviewname, TRIGGER_TYPE_UPDATE, TRIGGER_TYPE_BEFORE);
+ CreateIvmTrigger(rte->relid, matviewOid, matviewname, TRIGGER_TYPE_INSERT, TRIGGER_TYPE_AFTER);
+ CreateIvmTrigger(rte->relid, matviewOid, matviewname, TRIGGER_TYPE_DELETE, TRIGGER_TYPE_AFTER);
+ CreateIvmTrigger(rte->relid, matviewOid, matviewname, TRIGGER_TYPE_UPDATE, TRIGGER_TYPE_AFTER);
+
+ *relid_map = bms_add_member(*relid_map, rte->relid);
+ }
+ }
+ else
+ elog(ERROR, "unsupported RTE kind: %d", (int) rte->rtekind);
+ }
+ else if (IsA(jtnode, FromExpr))
+ {
+ FromExpr *f = (FromExpr *) jtnode;
+ ListCell *l;
+
+ foreach(l, f->fromlist)
+ CreateIvmTriggersOnBaseTables(qry, lfirst(l), matviewOid, matviewname, relid_map);
+ }
+ else if (IsA(jtnode, JoinExpr))
+ {
+ JoinExpr *j = (JoinExpr *) jtnode;
+
+ CreateIvmTriggersOnBaseTables(qry, j->larg, matviewOid, matviewname, relid_map);
+ CreateIvmTriggersOnBaseTables(qry, j->rarg, matviewOid, matviewname, relid_map);
+ }
+ else
+ elog(ERROR, "unrecognized node type: %d", (int) nodeTag(jtnode));
+}
+
/*
* GetIntoRelEFlags --- compute executor flags needed for CREATE TABLE AS
*
@@ -547,6 +776,11 @@ intorel_startup(DestReceiver *self, int operation, TupleDesc typeinfo)
if (is_matview && !into->skipData)
SetMatViewPopulatedState(intoRelationDesc, true);
+ /*
+ * Mark relisivm field, if it's a matview and into->ivm is true.
+ */
+ if (is_matview && into->ivm)
+ SetMatViewIVMState(intoRelationDesc, true);
/*
* Fill private fields of myState for use by later routines
*/
@@ -619,3 +853,237 @@ intorel_destroy(DestReceiver *self)
{
pfree(self);
}
+
+
+static void
+CreateIvmTrigger(Oid relOid, Oid viewOid, char *matviewname, int16 type, int16 timing)
+{
+ CreateTrigStmt *ivm_trigger;
+ List *transitionRels = NIL;
+ ObjectAddress address, refaddr;
+
+ Assert(timing == TRIGGER_TYPE_BEFORE || timing == TRIGGER_TYPE_AFTER);
+
+ refaddr.classId = RelationRelationId;
+ refaddr.objectId = viewOid;
+ refaddr.objectSubId = 0;
+
+ ivm_trigger = makeNode(CreateTrigStmt);
+ ivm_trigger->relation = NULL;
+ ivm_trigger->row = false;
+
+ ivm_trigger->timing = timing;
+ ivm_trigger->events = type;
+
+ switch (type)
+ {
+ case TRIGGER_TYPE_INSERT:
+ ivm_trigger->trigname = (timing == TRIGGER_TYPE_BEFORE ? "IVM_trigger_ins_before" : "IVM_trigger_ins_after");
+ break;
+ case TRIGGER_TYPE_DELETE:
+ ivm_trigger->trigname = (timing == TRIGGER_TYPE_BEFORE ? "IVM_trigger_del_before" : "IVM_trigger_del_after");
+ break;
+ case TRIGGER_TYPE_UPDATE:
+ ivm_trigger->trigname = (timing == TRIGGER_TYPE_BEFORE ? "IVM_trigger_upd_before" : "IVM_trigger_upd_after");
+ break;
+ default:
+ elog(ERROR, "unsupported trigger type");
+ }
+
+ if (timing == TRIGGER_TYPE_AFTER)
+ {
+ if (type == TRIGGER_TYPE_INSERT || type == TRIGGER_TYPE_UPDATE)
+ {
+ TriggerTransition *n = makeNode(TriggerTransition);
+ n->name = "ivm_newtable";
+ n->isNew = true;
+ n->isTable = true;
+
+ transitionRels = lappend(transitionRels, n);
+ }
+ if (type == TRIGGER_TYPE_DELETE || type == TRIGGER_TYPE_UPDATE)
+ {
+ TriggerTransition *n = makeNode(TriggerTransition);
+ n->name = "ivm_oldtable";
+ n->isNew = false;
+ n->isTable = true;
+
+ transitionRels = lappend(transitionRels, n);
+ }
+ }
+
+ ivm_trigger->funcname =
+ (timing == TRIGGER_TYPE_BEFORE ? SystemFuncName("IVM_immediate_before") : SystemFuncName("IVM_immediate_maintenance"));
+
+ ivm_trigger->columns = NIL;
+ ivm_trigger->transitionRels = transitionRels;
+ ivm_trigger->whenClause = NULL;
+ ivm_trigger->isconstraint = false;
+ ivm_trigger->deferrable = false;
+ ivm_trigger->initdeferred = false;
+ ivm_trigger->constrrel = NULL;
+ ivm_trigger->args = list_make1(makeString(matviewname));
+
+ address = CreateTrigger(ivm_trigger, NULL, relOid, InvalidOid, InvalidOid,
+ InvalidOid, InvalidOid, InvalidOid, NULL, true, false);
+
+ recordDependencyOn(&address, &refaddr, DEPENDENCY_AUTO);
+
+ /* Make changes-so-far visible */
+ CommandCounterIncrement();
+}
+
+/*
+ * check_ivm_restriction_walker --- look for specify nodes in the query tree
+ */
+static void
+check_ivm_restriction_walker(Node *node)
+{
+ /* This can recurse, so check for excessive recursion */
+ check_stack_depth();
+
+ if (node == NULL)
+ return;
+ /*
+ * We currently don't support Sub-Query.
+ */
+ if (IsA(node, SubPlan) || IsA(node, SubLink))
+ ereport(ERROR, (errmsg("subquery is not supported with IVM")));
+
+ switch (nodeTag(node))
+ {
+ case T_Query:
+ {
+ Query *qry = (Query *)node;
+ ListCell *lc;
+ /* if contained CTE, return error */
+ if (qry->cteList != NIL)
+ ereport(ERROR, (errmsg("CTE is not supported with IVM")));
+
+ /* if contained VIEW or subquery into RTE, return error */
+ foreach(lc, qry->rtable)
+ {
+ RangeTblEntry *rte = (RangeTblEntry *) lfirst(lc);
+ if (rte->relkind == RELKIND_VIEW ||
+ rte->relkind == RELKIND_MATVIEW)
+ ereport(ERROR, (errmsg("VIEW or MATERIALIZED VIEW is not supported with IVM")));
+ if (rte->rtekind == RTE_SUBQUERY)
+ ereport(ERROR, (errmsg("subquery is not supported with IVM")));
+ }
+
+ /* search in jointree */
+ check_ivm_restriction_walker((Node *) qry->jointree);
+
+ /* search in target lists */
+ foreach(lc, qry->targetList)
+ {
+ TargetEntry *tle = (TargetEntry *) lfirst(lc);
+ check_ivm_restriction_walker((Node *) tle->expr);
+ }
+
+ break;
+ }
+ case T_JoinExpr:
+ {
+ JoinExpr *joinexpr = (JoinExpr *)node;
+ if (joinexpr->jointype > JOIN_INNER)
+ ereport(ERROR, (errmsg("OUTER JOIN is not supported with IVM")));
+ /* left side */
+ check_ivm_restriction_walker((Node *) joinexpr->larg);
+ /* right side */
+ check_ivm_restriction_walker((Node *) joinexpr->rarg);
+ check_ivm_restriction_walker((Node *) joinexpr->quals);
+ }
+ break;
+ case T_FromExpr:
+ {
+ ListCell *lc;
+ FromExpr *fromexpr = (FromExpr *)node;
+ foreach(lc, fromexpr->fromlist)
+ {
+ check_ivm_restriction_walker((Node *) lfirst(lc));
+ }
+ check_ivm_restriction_walker((Node *) fromexpr->quals);
+ }
+ break;
+ case T_Var:
+ {
+ /* if system column, return error */
+ Var *variable = (Var *) node;
+ if (variable->varattno < 0)
+ ereport(ERROR, (errmsg("system column is not supported with IVM")));
+ }
+ break;
+ case T_BoolExpr:
+ {
+ BoolExpr *boolexpr = (BoolExpr *) node;
+ ListCell *lc;
+
+ foreach(lc, boolexpr->args)
+ {
+ Node *arg = (Node *) lfirst(lc);
+ check_ivm_restriction_walker(arg);
+ }
+ break;
+ }
+ case T_NullIfExpr: /* same as OpExpr */
+ case T_DistinctExpr: /* same as OpExpr */
+ case T_OpExpr:
+ {
+ OpExpr *op = (OpExpr *) node;
+ ListCell *lc;
+ foreach(lc, op->args)
+ {
+ Node *arg = (Node *) lfirst(lc);
+ check_ivm_restriction_walker(arg);
+ }
+ break;
+ }
+ case T_CaseExpr:
+ {
+ CaseExpr *caseexpr = (CaseExpr *) node;
+ ListCell *lc;
+ /* result for ELSE clause */
+ check_ivm_restriction_walker((Node *) caseexpr->defresult);
+ /* expr for WHEN clauses */
+ foreach(lc, caseexpr->args)
+ {
+ CaseWhen *when = (CaseWhen *) lfirst(lc);
+ Node *w_expr = (Node *) when->expr;
+ /* result for WHEN clause */
+ check_ivm_restriction_walker((Node *) when->result);
+ /* expr clause*/
+ check_ivm_restriction_walker((Node *) w_expr);
+ }
+ break;
+ }
+ case T_SubLink:
+ {
+ /* Now, not supported */
+/*
+ SubLink *sublink = (SubLink *) node;
+ Query *qry =(Query *) sublink->subselect;
+ if (qry != NULL && qry->jointree != NULL)
+ check_ivm_restriction_walker((Node *) qry->jointree->quals);
+*/
+ break;
+ }
+ case T_SubPlan:
+ {
+ /* Now, not supported */
+ break;
+ }
+ case T_Aggref:
+ case T_GroupingFunc:
+ case T_WindowFunc:
+ case T_FuncExpr:
+ case T_SQLValueFunction:
+ case T_Const:
+ case T_Param:
+ default:
+ /* do nothing */
+ break;
+ }
+
+ return;
+}
diff --git a/src/backend/commands/matview.c b/src/backend/commands/matview.c
index 537d0e8cef..c2831590b8 100644
--- a/src/backend/commands/matview.c
+++ b/src/backend/commands/matview.c
@@ -46,6 +46,79 @@
#include "utils/snapmgr.h"
#include "utils/syscache.h"
+#include "utils/regproc.h"
+#include "nodes/makefuncs.h"
+#include "parser/analyze.h"
+#include "parser/parser.h"
+#include "parser/parse_clause.h"
+#include "parser/parse_func.h"
+#include "parser/parse_type.h"
+#include "nodes/print.h"
+#include "catalog/pg_type_d.h"
+#include "optimizer/optimizer.h"
+#include "commands/defrem.h"
+
+/*
+ * Local definitions
+ */
+
+#define MV_INIT_QUERYHASHSIZE 16
+
+/* MV query type codes */
+#define MV_PLAN_RECALC_MINMAX 1
+#define MV_PLAN_SET_MINMAX 2
+
+/*
+ * MI_QueryKey
+ *
+ * The key identifying a prepared SPI plan in our query hashtable
+ */
+typedef struct MV_QueryKey
+{
+ Oid matview_id; /* OID of materialized view */
+ int32 query_type; /* query type ID, see MV_PLAN_XXX above */
+} MV_QueryKey;
+
+/*
+ * MV_QueryHashEntry
+ */
+typedef struct MV_QueryHashEntry
+{
+ MV_QueryKey key;
+ SPIPlanPtr plan;
+} MV_QueryHashEntry;
+
+/*
+ * MV_TriggerHashEntry
+ */
+typedef struct MV_TriggerHashEntry
+{
+ Oid matview_id;
+ int before_trig_count;
+ int after_trig_count;
+ TransactionId xid;
+ CommandId cid;
+ List *tables;
+ bool has_old;
+ bool has_new;
+} MV_TriggerHashEntry;
+
+/*
+ * MV_TiggerTable
+ */
+typedef struct MV_TriggerTable
+{
+ Oid table_id;
+ List *old_tuplestores;
+ List *new_tuplestores;
+ RangeTblEntry *original_rte;
+ List *old_rtes;
+ List *new_rtes;
+ List *rte_indexes;
+} MV_TriggerTable;
+
+static HTAB *mv_query_cache = NULL;
+static HTAB *mv_trigger_info = NULL;
typedef struct
{
@@ -58,6 +131,12 @@ typedef struct
BulkInsertState bistate; /* bulk insert state */
} DR_transientrel;
+typedef enum
+{
+ IVM_ADD,
+ IVM_SUB
+} IvmOp;
+
static int matview_maintenance_depth = 0;
static void transientrel_startup(DestReceiver *self, int operation, TupleDesc typeinfo);
@@ -65,7 +144,8 @@ 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);
+ QueryEnvironment *queryEnv,
+ const char *queryString);
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);
@@ -74,6 +154,33 @@ static bool is_usable_unique_index(Relation indexRel);
static void OpenMatViewIncrementalMaintenance(void);
static void CloseMatViewIncrementalMaintenance(void);
+static char *get_null_condition_string(IvmOp op, char *arg1, char *arg2, char* count_col);
+static char *get_operation_string(IvmOp op, char *col, char *arg1, char *arg2,
+ char* count_col, const char *castType);
+static Query* rewrite_query_for_preupdate_state(Query *query, List *tables, TransactionId xid, CommandId cid, ParseState *pstate);
+static Query *rewrite_query_for_counting_and_aggregation(Query *query, ParseState *pstate);
+static void calc_delta(MV_TriggerTable *table, int index, Query *query,
+ DestReceiver *dest_old, DestReceiver *dest_new, QueryEnvironment *queryEnv);
+static RangeTblEntry* union_ENRs(RangeTblEntry *rte, Oid relid, List *enr_rtes, const char *prefix, QueryEnvironment *queryEnv);
+static char *make_delta_enr_name(const char *prefix, Oid relid, int count);
+static void register_delta_ENRs(ParseState *pstate, Query *query, List *tables);
+static void apply_delta(Oid matviewOid, Oid tempOid_new, Oid tempOid_old, Query *query);
+static void truncate_view_delta(Oid delta_oid);
+static void clean_up_IVM_hash_entry(MV_TriggerHashEntry *entry);
+static void clean_up_IVM_temptable(Oid tempOid_old, Oid tempOid_new);
+
+static void mv_InitHashTables(void);
+static SPIPlanPtr mv_FetchPreparedPlan(MV_QueryKey *key);
+static void mv_HashPreparedPlan(MV_QueryKey *key, SPIPlanPtr plan);
+static void mv_BuildQueryKey(MV_QueryKey *key, Oid matview_id, int32 query_type);
+static SPIPlanPtr get_plan_for_recalc_min_max(Oid matviewOid, const char *min_max_list,
+ const char *group_keys, int nkeys, Oid *keyTypes, bool with_group);
+static SPIPlanPtr get_plan_for_set_min_max(Oid matviewOid, char *matviewname, const char *min_max_list,
+ int num_min_max, Oid *valTypes, bool with_group);
+
+static Query *get_matview_query(Relation matviewRel);
+
+
/*
* SetMatViewPopulatedState
* Mark a materialized view as populated, or not.
@@ -114,6 +221,46 @@ SetMatViewPopulatedState(Relation relation, bool newstate)
CommandCounterIncrement();
}
+/*
+ * SetMatViewIVMState
+ * Mark a materialized view as IVM, or not.
+ *
+ * NOTE: caller must be holding an appropriate lock on the relation.
+ */
+void
+SetMatViewIVMState(Relation relation, bool newstate)
+{
+ Relation pgrel;
+ HeapTuple tuple;
+
+ Assert(relation->rd_rel->relkind == RELKIND_MATVIEW);
+
+ /*
+ * Update relation's pg_class entry. Crucial side-effect: other backends
+ * (and this one too!) are sent SI message to make them rebuild relcache
+ * entries.
+ */
+ pgrel = table_open(RelationRelationId, RowExclusiveLock);
+ tuple = SearchSysCacheCopy1(RELOID,
+ ObjectIdGetDatum(RelationGetRelid(relation)));
+ if (!HeapTupleIsValid(tuple))
+ elog(ERROR, "cache lookup failed for relation %u",
+ RelationGetRelid(relation));
+
+ ((Form_pg_class) GETSTRUCT(tuple))->relisivm = newstate;
+
+ CatalogTupleUpdate(pgrel, &tuple->t_self, tuple);
+
+ heap_freetuple(tuple);
+ table_close(pgrel, RowExclusiveLock);
+
+ /*
+ * Advance command counter to make the updated pg_class row locally
+ * visible.
+ */
+ CommandCounterIncrement();
+}
+
/*
* ExecRefreshMatView -- execute a REFRESH MATERIALIZED VIEW command
*
@@ -140,8 +287,6 @@ ExecRefreshMatView(RefreshMatViewStmt *stmt, const char *queryString,
{
Oid matviewOid;
Relation matviewRel;
- RewriteRule *rule;
- List *actions;
Query *dataQuery;
Oid tableSpace;
Oid relowner;
@@ -187,32 +332,8 @@ ExecRefreshMatView(RefreshMatViewStmt *stmt, const char *queryString,
(errcode(ERRCODE_SYNTAX_ERROR),
errmsg("CONCURRENTLY and WITH NO DATA options cannot be used together")));
- /*
- * Check that everything is correct for a refresh. Problems at this point
- * are internal errors, so elog is sufficient.
- */
- if (matviewRel->rd_rel->relhasrules == false ||
- matviewRel->rd_rules->numLocks < 1)
- elog(ERROR,
- "materialized view \"%s\" is missing rewrite information",
- RelationGetRelationName(matviewRel));
-
- if (matviewRel->rd_rules->numLocks > 1)
- elog(ERROR,
- "materialized view \"%s\" has too many rules",
- RelationGetRelationName(matviewRel));
-
- rule = matviewRel->rd_rules->rules[0];
- if (rule->event != CMD_SELECT || !(rule->isInstead))
- elog(ERROR,
- "the rule for materialized view \"%s\" is not a SELECT INSTEAD OF rule",
- RelationGetRelationName(matviewRel));
- actions = rule->actions;
- if (list_length(actions) != 1)
- elog(ERROR,
- "the rule for materialized view \"%s\" is not a single action",
- RelationGetRelationName(matviewRel));
+ dataQuery = get_matview_query(matviewRel);
/*
* Check that there is a unique index with no WHERE clause on one or more
@@ -247,12 +368,6 @@ ExecRefreshMatView(RefreshMatViewStmt *stmt, const char *queryString,
errhint("Create a unique index with no WHERE clause on one or more columns of the materialized view.")));
}
- /*
- * The stored query was rewritten at the time of the MV definition, but
- * has not been scribbled on by the planner.
- */
- dataQuery = linitial_node(Query, actions);
-
/*
* Check for active uses of the relation in the current transaction, such
* as open scans.
@@ -311,7 +426,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, NULL, queryString);
/* Make the matview match the newly generated data. */
if (concurrent)
@@ -369,6 +484,7 @@ ExecRefreshMatView(RefreshMatViewStmt *stmt, const char *queryString,
*/
static uint64
refresh_matview_datafill(DestReceiver *dest, Query *query,
+ QueryEnvironment *queryEnv,
const char *queryString)
{
List *rewritten;
@@ -405,7 +521,7 @@ refresh_matview_datafill(DestReceiver *dest, Query *query,
/* Create a QueryDesc, redirecting output to our tuple receiver */
queryDesc = CreateQueryDesc(plan, queryString,
GetActiveSnapshot(), InvalidSnapshot,
- dest, NULL, NULL, 0);
+ dest, NULL, queryEnv ? queryEnv: NULL, 0);
/* call ExecutorStart to prepare the plan for execution */
ExecutorStart(queryDesc, 0);
@@ -926,3 +1042,1580 @@ CloseMatViewIncrementalMaintenance(void)
matview_maintenance_depth--;
Assert(matview_maintenance_depth >= 0);
}
+
+/*
+ * IVM trigger function
+ */
+
+Datum
+IVM_immediate_before(PG_FUNCTION_ARGS)
+{
+ TriggerData *trigdata = (TriggerData *) fcinfo->context;
+ char *matviewname = trigdata->tg_trigger->tgargs[0];
+ List *names = stringToQualifiedNameList(matviewname);
+ Oid matviewOid;
+
+ MV_TriggerHashEntry *entry;
+ bool found;
+
+ /*
+ * Wait for concurrent transactions which update this materialized view at READ COMMITED.
+ * This is needed to see changes commited in othre transactions. No wait and raise an error
+ * at REPEATABLE READ or SERIALIZABLE to prevent anormal update of matviews.
+ * XXX: dead-lock is possible here.
+ */
+ if (!IsolationUsesXactSnapshot())
+ matviewOid = RangeVarGetRelid(makeRangeVarFromNameList(names), ExclusiveLock, true);
+ else
+ matviewOid = RangeVarGetRelidExtended(makeRangeVarFromNameList(names), ExclusiveLock, RVR_MISSING_OK | RVR_NOWAIT, NULL, NULL);
+
+ /*
+ * On the first call initialize the hashtable
+ */
+ if (!mv_query_cache)
+ mv_InitHashTables();
+
+ entry = (MV_TriggerHashEntry *) hash_search(mv_trigger_info,
+ (void *) &matviewOid,
+ HASH_ENTER, &found);
+
+ /* On the first BEFORE to update the view, initialize trigger data */
+ if (!found)
+ {
+ Snapshot snapshot = GetActiveSnapshot();
+
+ entry->matview_id = matviewOid;
+ entry->before_trig_count = 0;
+ entry->after_trig_count = 0;
+ entry->xid = GetCurrentTransactionId();
+ entry->cid = snapshot->curcid;
+ entry->tables = NIL;
+ entry->has_old = false;
+ entry->has_new = false;
+
+ }
+
+ entry->before_trig_count++;
+
+ return PointerGetDatum(NULL);
+}
+
+Datum
+IVM_immediate_maintenance(PG_FUNCTION_ARGS)
+{
+ TriggerData *trigdata = (TriggerData *) fcinfo->context;
+ Relation rel;
+ Oid relid;
+ Oid matviewOid;
+ Query *query, *rewritten;
+ char* matviewname = trigdata->tg_trigger->tgargs[0];
+ List *names;
+ Relation matviewRel;
+ int old_depth = matview_maintenance_depth;
+
+ Oid tableSpace;
+ Oid relowner;
+ Oid OIDDelta_new = InvalidOid;
+ Oid OIDDelta_old = InvalidOid;
+ DestReceiver *dest_new = NULL, *dest_old = NULL;
+ Oid save_userid;
+ int save_sec_context;
+ int save_nestlevel;
+
+ MV_TriggerHashEntry *entry;
+ MV_TriggerTable *table;
+ bool found;
+ ListCell *lc;
+ MemoryContext oldcxt;
+
+
+ QueryEnvironment *queryEnv = create_queryEnv();
+ ParseState *pstate;
+
+ /* Create a ParseState for rewriting the view definition query */
+ pstate = make_parsestate(NULL);
+ pstate->p_queryEnv = queryEnv;
+ pstate->p_expr_kind = EXPR_KIND_SELECT_TARGET;
+
+ rel = trigdata->tg_relation;
+ relid = rel->rd_id;
+
+ names = stringToQualifiedNameList(matviewname);
+ matviewOid = RangeVarGetRelid(makeRangeVarFromNameList(names), NoLock, true);
+
+
+ /*
+ * On the first call initialize the hashtable
+ */
+ if (!mv_query_cache)
+ mv_InitHashTables();
+
+ entry = (MV_TriggerHashEntry *) hash_search(mv_trigger_info,
+ (void *) &matviewOid,
+ HASH_FIND, &found);
+
+ Assert (entry != NULL);
+
+ entry->after_trig_count++;
+
+ found = false;
+ foreach(lc, entry->tables)
+ {
+ table = (MV_TriggerTable *) lfirst(lc);
+ if (table->table_id == relid)
+ {
+ found = true;
+ break;
+ }
+ }
+ if (!found)
+ {
+ oldcxt = MemoryContextSwitchTo(CurTransactionContext);
+
+ table = (MV_TriggerTable *) palloc0(sizeof(MV_TriggerTable));
+ table->table_id = relid;
+ table->old_tuplestores = NIL;
+ table->new_tuplestores = NIL;
+ table->old_rtes = NIL;
+ table->new_rtes = NIL;
+ table->rte_indexes = NIL;
+ entry->tables = lappend(entry->tables, table);
+
+ MemoryContextSwitchTo(oldcxt);
+ }
+
+ if (trigdata->tg_oldtable)
+ {
+ oldcxt = MemoryContextSwitchTo(CurTransactionContext);
+ table->old_tuplestores = lappend(table->old_tuplestores, trigdata->tg_oldtable);
+ entry->has_old = true;
+ MemoryContextSwitchTo(oldcxt);
+ }
+ if (trigdata->tg_newtable)
+ {
+ oldcxt = MemoryContextSwitchTo(CurTransactionContext);
+ table->new_tuplestores = lappend(table->new_tuplestores, trigdata->tg_newtable);
+ entry->has_new = true;
+ MemoryContextSwitchTo(oldcxt);
+ }
+ if (entry->has_new || entry->has_old)
+ {
+ CmdType cmd;
+
+ if (TRIGGER_FIRED_BY_INSERT(trigdata->tg_event))
+ cmd = CMD_INSERT;
+ else if (TRIGGER_FIRED_BY_DELETE(trigdata->tg_event))
+ cmd = CMD_DELETE;
+ else if (TRIGGER_FIRED_BY_UPDATE(trigdata->tg_event))
+ cmd = CMD_UPDATE;
+ else
+ elog(ERROR,"unsupported trigger type");
+
+ SetTransitionTablePreserved(relid, cmd);
+ }
+
+
+ /* If this is not the last AFTER trigger call, immediately exit. */
+ Assert (entry->before_trig_count >= entry->after_trig_count);
+ if (entry->before_trig_count != entry->after_trig_count)
+ return PointerGetDatum(NULL);
+
+
+ /* If this is the last AFTER trigger call, update the view. */
+
+ matviewRel = table_open(matviewOid, NoLock);
+
+ /* get view query*/
+ query = get_matview_query(matviewRel);
+
+ /* Make sure it is a materialized view. */
+ Assert(matviewRel->rd_rel->relkind == RELKIND_MATVIEW);
+
+ /*
+ * Get and push the latast snapshot to see any changes which is commited during waiting in
+ * other transactions at READ COMMITTED level.
+ * XXX: Is this safe?
+ */
+ PushActiveSnapshot(GetTransactionSnapshot());
+
+ /*
+ * Check for active uses of the relation in the current transaction, such
+ * as open scans.
+ *
+ * NB: We count on this to protect us against problems with refreshing the
+ * data using TABLE_INSERT_FROZEN.
+ */
+ // XXX: necesarry?
+ CheckTableNotInUse(matviewRel, "REFRESH MATERIALIZED VIEW");
+
+ /* rewrite query */
+ rewritten = rewrite_query_for_preupdate_state(query, entry->tables, entry->xid, entry->cid, pstate);
+ rewritten = rewrite_query_for_counting_and_aggregation(rewritten, pstate);
+
+ relowner = matviewRel->rd_rel->relowner;
+
+ /*
+ * Switch to the owner's userid, so that any functions are run as that
+ * user. Also arrange to make GUC variable changes local to this command.
+ * Don't lock it down too tight to create a temporary table just yet. We
+ * will switch modes when we are about to execute user code.
+ */
+ GetUserIdAndSecContext(&save_userid, &save_sec_context);
+ SetUserIdAndSecContext(relowner,
+ save_sec_context | SECURITY_LOCAL_USERID_CHANGE);
+ save_nestlevel = NewGUCNestLevel();
+
+ /* Create temporary tables to store view deltas */
+ tableSpace = GetDefaultTablespace(RELPERSISTENCE_TEMP, false);
+ if (entry->has_old)
+ {
+ OIDDelta_old = make_new_heap(matviewOid, tableSpace, RELPERSISTENCE_TEMP,
+ ExclusiveLock);
+ LockRelationOid(OIDDelta_old, AccessExclusiveLock);
+ dest_old = CreateTransientRelDestReceiver(OIDDelta_old);
+ }
+ if (entry->has_new)
+ {
+ if (entry->has_old)
+ OIDDelta_new = make_new_heap(OIDDelta_old, tableSpace, RELPERSISTENCE_TEMP,
+ ExclusiveLock);
+ else
+ OIDDelta_new = make_new_heap(matviewOid, tableSpace, RELPERSISTENCE_TEMP,
+ ExclusiveLock);
+ LockRelationOid(OIDDelta_new, AccessExclusiveLock);
+ dest_new = CreateTransientRelDestReceiver(OIDDelta_new);
+ }
+
+ /*
+ * Now lock down security-restricted operations.
+ */
+ SetUserIdAndSecContext(relowner,
+ save_sec_context | SECURITY_RESTRICTED_OPERATION);
+
+ /* for all modified tables */
+ foreach(lc, entry->tables)
+ {
+ ListCell *lc2;
+
+ table = (MV_TriggerTable *) lfirst(lc);
+
+ /* loop for self-join */
+ foreach(lc2, table->rte_indexes)
+ {
+ int index = lfirst_int(lc2);
+
+ calc_delta(table, index, rewritten, dest_old, dest_new, queryEnv);
+
+ PG_TRY();
+ {
+ apply_delta(matviewOid, OIDDelta_new, OIDDelta_old, query);
+ }
+ PG_CATCH();
+ {
+ matview_maintenance_depth = old_depth;
+ PG_RE_THROW();
+ }
+ PG_END_TRY();
+
+ /* truncate view delta tables */
+ truncate_view_delta(OIDDelta_old);
+ truncate_view_delta(OIDDelta_new);
+ }
+ }
+
+ /* Pop the original snapshot. */
+ PopActiveSnapshot();
+
+ table_close(matviewRel, NoLock);
+
+ /* Roll back any GUC changes */
+ AtEOXact_GUC(false, save_nestlevel);
+
+ /* Restore userid and security context */
+ SetUserIdAndSecContext(save_userid, save_sec_context);
+
+ /* Clean up hash entry and drop temporary tables */
+ clean_up_IVM_hash_entry(entry);
+ clean_up_IVM_temptable(OIDDelta_old, OIDDelta_new);
+
+ return PointerGetDatum(NULL);
+}
+
+#define IVM_colname(type, col) makeObjectName("__ivm_" type, col, "_")
+
+static char *
+get_null_condition_string(IvmOp op, char *arg1, char *arg2, char* count_col)
+{
+ StringInfoData null_cond;
+ initStringInfo(&null_cond);
+
+ switch (op)
+ {
+ case IVM_ADD:
+ appendStringInfo(&null_cond,
+ "%s = 0 AND %s = 0",
+ quote_qualified_identifier(arg1, count_col),
+ quote_qualified_identifier(arg2, count_col)
+ );
+ break;
+ case IVM_SUB:
+ appendStringInfo(&null_cond,
+ "%s = %s",
+ quote_qualified_identifier(arg1, count_col),
+ quote_qualified_identifier(arg2, count_col)
+ );
+ break;
+ default:
+ elog(ERROR,"unkwon opration");
+ }
+
+ return null_cond.data;
+}
+
+static char *
+get_operation_string(IvmOp op, char *col, char *arg1, char *arg2, char* count_col, const char *castType)
+{
+ StringInfoData buf, castString;
+ char *col1 = quote_qualified_identifier(arg1, col);
+ char *col2 = quote_qualified_identifier(arg2, col);
+
+ char op_char = (op == IVM_SUB ? '-' : '+');
+ char *sign = (op == IVM_SUB ? "-" : "");
+
+ initStringInfo(&buf);
+ initStringInfo(&castString);
+
+ if (castType)
+ appendStringInfo(&castString, "::%s", castType);
+
+
+ if (!count_col)
+ {
+ appendStringInfo(&buf, "(%s %c %s)%s",
+ col1, op_char, col2, castString.data);
+ }
+ else
+ {
+ char *null_cond = get_null_condition_string(op, arg1, arg2, count_col);
+
+ appendStringInfo(&buf,
+ "(CASE WHEN %s THEN NULL "
+ " WHEN %s IS NULL THEN %s%s "
+ " WHEN %s IS NULL THEN (%s) "
+ " ELSE (%s %c %s)%s END)",
+ null_cond,
+ col1, sign, col2,
+ col2, col1,
+ col1, op_char, col2, castString.data
+ );
+ }
+
+ return buf.data;
+}
+
+
+static void
+calc_delta(MV_TriggerTable *table, int index, Query *query,
+ DestReceiver *dest_old, DestReceiver *dest_new, QueryEnvironment *queryEnv)
+{
+ ListCell *lc;
+ RangeTblEntry *rte;
+
+ lc = list_nth_cell(query->rtable, index);
+ rte = (RangeTblEntry *) lfirst(lc);
+
+ /* Generate old delta */
+ if (list_length(table->old_rtes) > 0)
+ {
+ lfirst(lc) = union_ENRs(rte, table->table_id, table->old_rtes, "old", queryEnv);
+ refresh_matview_datafill(dest_old, query, queryEnv, NULL);
+ }
+
+ /* Generate new delta */
+ if (list_length(table->new_rtes) > 0)
+ {
+ lfirst(lc) = union_ENRs(rte, table->table_id, table->new_rtes, "new", queryEnv);
+ refresh_matview_datafill(dest_new, query, queryEnv, NULL);
+ }
+
+ /* Retore the original RTE */
+ lfirst(lc) = table->original_rte;
+}
+
+
+static RangeTblEntry*
+union_ENRs(RangeTblEntry *rte, Oid relid, List *enr_rtes, const char *prefix, QueryEnvironment *queryEnv)
+{
+ StringInfoData str;
+ ParseState *pstate;
+ RawStmt *raw;
+ Query *sub;
+ int i;
+
+ /* Create a ParseState for rewriting the view definition query */
+ pstate = make_parsestate(NULL);
+ pstate->p_queryEnv = queryEnv;
+ pstate->p_expr_kind = EXPR_KIND_SELECT_TARGET;
+
+ initStringInfo(&str);
+
+ for (i = 0; i < list_length(enr_rtes); i++)
+ {
+ if (i > 0)
+ appendStringInfo(&str, " UNION ALL ");
+ appendStringInfo(&str," SELECT * FROM %s", make_delta_enr_name(prefix, relid, i));
+ }
+
+ raw = (RawStmt*)linitial(raw_parser(str.data));
+ sub = transformStmt(pstate, raw->stmt);
+
+ rte->rtekind = RTE_SUBQUERY;
+ rte->subquery = sub;
+ rte->security_barrier = false;
+ /* Clear fields that should not be set in a subquery RTE */
+ rte->relid = InvalidOid;
+ rte->relkind = 0;
+ rte->rellockmode = 0;
+ rte->tablesample = NULL;
+ rte->inh = false; /* must not be set for a subquery */
+
+ rte->requiredPerms = 0; /* no permission check on subquery itself */
+ rte->checkAsUser = InvalidOid;
+ rte->selectedCols = NULL;
+ rte->insertedCols = NULL;
+ rte->updatedCols = NULL;
+ rte->extraUpdatedCols = NULL;
+
+ return rte;
+}
+
+static char*
+make_delta_enr_name(const char *prefix, Oid relid, int count)
+{
+ char buf[NAMEDATALEN];
+ char *name;
+
+ snprintf(buf, NAMEDATALEN, "%s_%u_%u", prefix, relid, count);
+ name = pstrdup(buf);
+
+ return name;
+}
+
+static void
+register_delta_ENRs(ParseState *pstate, Query *query, List *tables)
+{
+ QueryEnvironment *queryEnv = pstate->p_queryEnv;
+ ListCell *lc;
+ RangeTblEntry *rte;
+
+ foreach(lc, tables)
+ {
+ MV_TriggerTable *table = (MV_TriggerTable *) lfirst(lc);
+ ListCell *lc2;
+ int count;
+
+ count = 0;
+ foreach(lc2, table->old_tuplestores)
+ {
+ Tuplestorestate *oldtable = (Tuplestorestate *) lfirst(lc2);
+ EphemeralNamedRelation enr =
+ palloc(sizeof(EphemeralNamedRelationData));
+
+ enr->md.name = make_delta_enr_name("old", table->table_id, count);
+ enr->md.reliddesc = table->table_id;
+ enr->md.tupdesc = NULL;
+ enr->md.enrtype = ENR_NAMED_TUPLESTORE;
+ enr->md.enrtuples = tuplestore_tuple_count(oldtable);
+ enr->reldata = oldtable;
+ register_ENR(queryEnv, enr);
+
+ rte = addRangeTableEntryForENR(pstate, makeRangeVar(NULL, enr->md.name, -1), true);
+ query->rtable = lappend(query->rtable, rte);
+ table->old_rtes = lappend(table->old_rtes, rte);
+
+ count++;
+ }
+
+ count = 0;
+ foreach(lc2, table->new_tuplestores)
+ {
+ Tuplestorestate *newtable = (Tuplestorestate *) lfirst(lc2);
+ EphemeralNamedRelation enr =
+ palloc(sizeof(EphemeralNamedRelationData));
+
+ enr->md.name = make_delta_enr_name("new", table->table_id, count);
+ enr->md.reliddesc = table->table_id;
+ enr->md.tupdesc = NULL;
+ enr->md.enrtype = ENR_NAMED_TUPLESTORE;
+ enr->md.enrtuples = tuplestore_tuple_count(newtable);
+ enr->reldata = newtable;
+ register_ENR(queryEnv, enr);
+
+ rte = addRangeTableEntryForENR(pstate, makeRangeVar(NULL, enr->md.name, -1), true);
+ query->rtable = lappend(query->rtable, rte);
+ table->new_rtes = lappend(table->new_rtes, rte);
+
+ count++;
+ }
+ }
+}
+
+static RangeTblEntry*
+get_prestate_rte(RangeTblEntry *rte, MV_TriggerTable *table, TransactionId xid, CommandId cid, QueryEnvironment *queryEnv)
+{
+ StringInfoData str;
+ RawStmt *raw;
+ Query *sub;
+ Relation rel;
+ ParseState *pstate;
+ char *relname;
+ int i;
+
+ pstate = make_parsestate(NULL);
+ pstate->p_queryEnv = queryEnv;
+ pstate->p_expr_kind = EXPR_KIND_SELECT_TARGET;
+
+ /*
+ * We can use NoLock here since AcquireRewriteLocks should
+ * have locked the rel already.
+ */
+ rel = table_open(table->table_id, NoLock);
+ relname = quote_qualified_identifier(
+ get_namespace_name(RelationGetNamespace(rel)),
+ RelationGetRelationName(rel));
+ table_close(rel, NoLock);
+
+ initStringInfo(&str);
+ appendStringInfo(&str,
+ "SELECT t.* FROM %s t"
+ " WHERE (age(t.xmin) - age(%u::text::xid) > 0) OR"
+ " (t.xmin = %u AND t.cmin::text::int < %u)",
+ relname, xid, xid, cid);
+
+ for (i=0; i<list_length(table->old_tuplestores); i++)
+ {
+ appendStringInfo(&str, " UNION ALL ");
+ appendStringInfo(&str," SELECT * FROM %s", make_delta_enr_name("old", table->table_id, i));
+ }
+
+ raw = (RawStmt*)linitial(raw_parser(str.data));
+ sub = transformStmt(pstate, raw->stmt);
+
+ table->original_rte = copyObject(rte);
+
+ rte->rtekind = RTE_SUBQUERY;
+ rte->subquery = sub;
+ rte->security_barrier = false;
+ /* Clear fields that should not be set in a subquery RTE */
+ rte->relid = InvalidOid;
+ rte->relkind = 0;
+ rte->rellockmode = 0;
+ rte->tablesample = NULL;
+ rte->inh = false; /* must not be set for a subquery */
+
+ rte->requiredPerms = 0; /* no permission check on subquery itself */
+ rte->checkAsUser = InvalidOid;
+ rte->selectedCols = NULL;
+ rte->insertedCols = NULL;
+ rte->updatedCols = NULL;
+ rte->extraUpdatedCols = NULL;
+
+ return rte;
+}
+
+static Query*
+rewrite_query_for_preupdate_state(Query *query, List *tables, TransactionId xid, CommandId cid, ParseState *pstate)
+{
+ Query *rewritten = copyObject(query);
+ ListCell *lc;
+ int num_rte = list_length(query->rtable);
+ int i;
+
+ register_delta_ENRs(pstate, rewritten, tables);
+
+ // XXX: Is necessary? Is this right timing?
+ AcquireRewriteLocks(rewritten, true, false);
+
+ foreach(lc, tables)
+ {
+ MV_TriggerTable *table = (MV_TriggerTable *) lfirst(lc);
+ ListCell *lc2;
+
+ i = 0;
+ foreach(lc2, rewritten->rtable)
+ {
+ RangeTblEntry *r = (RangeTblEntry*) lfirst(lc2);
+
+ if (r->relid == table->table_id)
+ {
+ lfirst(lc2) = get_prestate_rte(r, table, xid, cid, pstate->p_queryEnv);
+ table->rte_indexes = lappend_int(table->rte_indexes, i);
+ }
+
+ if (++i >= num_rte)
+ break;
+ }
+ }
+
+ return rewritten;
+}
+
+static Query *
+rewrite_query_for_counting_and_aggregation(Query *query, ParseState *pstate)
+{
+ TargetEntry *tle_count;
+ FuncCall *fn;
+ Node *node;
+ Const *dmy_arg = makeConst(INT4OID,
+ -1,
+ InvalidOid,
+ sizeof(int32),
+ Int32GetDatum(1),
+ false,
+ true);
+
+ if (query->hasAggs)
+ {
+ ListCell *lc;
+ List *agg_counts = NIL;
+ AttrNumber next_resno = list_length(query->targetList) + 1;
+
+ foreach(lc, query->targetList)
+ {
+ TargetEntry *tle = (TargetEntry *) lfirst(lc);
+
+ if (IsA(tle->expr, Aggref))
+ {
+ Aggref *aggref = (Aggref *) tle->expr;
+ const char *aggname = get_func_name(aggref->aggfnoid);
+
+ /*
+ * For aggregate functions except to count, add count func with the same arg parameters.
+ * Also, add sum func for agv.
+ *
+ * XXX: need some generalization
+ * XXX: If there are same expressions explicitly in the target list, we can use this instead
+ * of adding new duplicated one.
+ */
+ if (strcmp(aggname, "count") != 0)
+ {
+ fn = makeFuncCall(list_make1(makeString("count")), NIL, -1);
+
+ /* Make a Func with a dummy arg, and then override this by the original agg's args. */
+ node = ParseFuncOrColumn(pstate, fn->funcname, list_make1(dmy_arg), NULL, fn, false, -1);
+ ((Aggref *)node)->args = aggref->args;
+
+ tle_count = makeTargetEntry((Expr *) node,
+ next_resno,
+ NULL,
+ false);
+ agg_counts = lappend(agg_counts, tle_count);
+ next_resno++;
+ }
+ if (strcmp(aggname, "avg") == 0)
+ {
+ List *dmy_args = NIL;
+ ListCell *lc;
+ foreach(lc, aggref->aggargtypes)
+ {
+ Oid typeid = lfirst_oid(lc);
+ Type type = typeidType(typeid);
+
+ Const *con = makeConst(typeid,
+ -1,
+ typeTypeCollation(type),
+ typeLen(type),
+ (Datum) 0,
+ true,
+ typeByVal(type));
+ dmy_args = lappend(dmy_args, con);
+ ReleaseSysCache(type);
+
+ }
+ fn = makeFuncCall(list_make1(makeString("sum")), NIL, -1);
+
+ /* Make a Func with a dummy arg, and then override this by the original agg's args. */
+ node = ParseFuncOrColumn(pstate, fn->funcname, dmy_args, NULL, fn, false, -1);
+ ((Aggref *)node)->args = aggref->args;
+
+ tle_count = makeTargetEntry((Expr *) node,
+ next_resno,
+ NULL,
+ false);
+ agg_counts = lappend(agg_counts, tle_count);
+ next_resno++;
+ }
+ }
+
+ }
+ query->targetList = list_concat(query->targetList, agg_counts);
+ }
+
+ fn = makeFuncCall(list_make1(makeString("count")), NIL, -1);
+ fn->agg_star = true;
+ if (!query->groupClause && !query->hasAggs)
+ query->groupClause = transformDistinctClause(NULL, &query->targetList, query->sortClause, false);
+
+ node = ParseFuncOrColumn(pstate, fn->funcname, NIL, NULL, fn, false, -1);
+
+ tle_count = makeTargetEntry((Expr *) node,
+ list_length(query->targetList) + 1,
+ NULL,
+ false);
+ query->targetList = lappend(query->targetList, tle_count);
+ query->hasAggs = true;
+
+ return query;
+}
+
+
+static void
+apply_delta(Oid matviewOid, Oid tempOid_new, Oid tempOid_old, Query *query)
+{
+ StringInfoData querybuf;
+ StringInfoData mvatts_buf, diffatts_buf;
+ StringInfoData mv_gkeys_buf, diff_gkeys_buf, updt_gkeys_buf;
+ StringInfoData diff_aggs_buf, update_aggs_old, update_aggs_new;
+ StringInfoData returning_buf, result_buf;
+ StringInfoData min_or_max_buf;
+ Relation matviewRel;
+ Relation tempRel_new = NULL, tempRel_old = NULL;
+ char *matviewname;
+ char *tempname_new = NULL, *tempname_old = NULL;
+ ListCell *lc;
+ char *sep, *sep_agg;
+ bool with_group = query->groupClause != NULL;
+ int i;
+ bool has_min_or_max = false;
+ int num_group_keys = 0;
+ int num_min_or_max = 0;
+
+
+ initStringInfo(&querybuf);
+ matviewRel = table_open(matviewOid, NoLock);
+ matviewname = quote_qualified_identifier(get_namespace_name(RelationGetNamespace(matviewRel)),
+ RelationGetRelationName(matviewRel));
+
+ if (OidIsValid(tempOid_new))
+ {
+ tempRel_new = table_open(tempOid_new, NoLock);
+ tempname_new = quote_qualified_identifier(get_namespace_name(RelationGetNamespace(tempRel_new)),
+ RelationGetRelationName(tempRel_new));
+ }
+ if (OidIsValid(tempOid_old))
+ {
+ tempRel_old = table_open(tempOid_old, NoLock);
+ tempname_old = quote_qualified_identifier(get_namespace_name(RelationGetNamespace(tempRel_old)),
+ RelationGetRelationName(tempRel_old));
+ }
+
+ initStringInfo(&mvatts_buf);
+ initStringInfo(&diffatts_buf);
+ initStringInfo(&diff_aggs_buf);
+ initStringInfo(&update_aggs_old);
+ initStringInfo(&update_aggs_new);
+ initStringInfo(&returning_buf);
+ initStringInfo(&result_buf);
+ initStringInfo(&min_or_max_buf);
+
+ sep = "";
+ sep_agg= "";
+ i = 0;
+ foreach (lc, query->targetList)
+ {
+ TargetEntry *tle = (TargetEntry *) lfirst(lc);
+ Form_pg_attribute attr = TupleDescAttr(matviewRel->rd_att, i);
+ char *resname = NameStr(attr->attname);
+
+ i++;
+
+ if (tle->resjunk)
+ continue;
+
+ appendStringInfo(&mvatts_buf, "%s", sep);
+ appendStringInfo(&diffatts_buf, "%s", sep);
+ sep = ", ";
+
+ appendStringInfo(&mvatts_buf, "%s", quote_qualified_identifier("mv", resname));
+ appendStringInfo(&diffatts_buf, "%s", quote_qualified_identifier("diff", resname));
+
+ if (query->hasAggs && IsA(tle->expr, Aggref))
+ {
+ Aggref *aggref = (Aggref *) tle->expr;
+ const char *aggname = get_func_name(aggref->aggfnoid);
+ const char *aggtype = format_type_be(aggref->aggtype); /* XXX: should be add_cast_to ? */
+
+ appendStringInfo(&update_aggs_old, "%s", sep_agg);
+ appendStringInfo(&update_aggs_new, "%s", sep_agg);
+ appendStringInfo(&diff_aggs_buf, "%s", sep_agg);
+
+ sep_agg = ", ";
+
+ /* XXX: need some generalization
+ *
+ * Specifically, Using func names is not robust. We can use oids instead
+ * of names, but it would be nice to add some information to pg_aggregate
+ * and handler functions.
+ */
+
+ if (!strcmp(aggname, "count"))
+ {
+ /* resname = mv.resname - t.resname */
+ appendStringInfo(&update_aggs_old,
+ "%s = %s",
+ quote_qualified_identifier(NULL, resname),
+ get_operation_string(IVM_SUB,resname, "mv", "t", NULL, NULL));
+
+ /* resname = mv.resname + diff.resname */
+ appendStringInfo(&update_aggs_new,
+ "%s = %s",
+ quote_qualified_identifier(NULL, resname),
+ get_operation_string(IVM_ADD, resname, "mv", "diff", NULL, NULL));
+
+ appendStringInfo(&diff_aggs_buf, "%s",
+ quote_qualified_identifier("diff", resname)
+ );
+ }
+ else if (!strcmp(aggname, "sum"))
+ {
+ char *count_col = IVM_colname("count", resname);
+
+ /* sum = mv.sum - t.sum */
+ appendStringInfo(&update_aggs_old,
+ "%s = %s, ",
+ quote_qualified_identifier(NULL, resname),
+ get_operation_string(IVM_SUB, resname, "mv", "t", count_col, NULL)
+ );
+ /* count = mv.count - t.count */
+ appendStringInfo(&update_aggs_old,
+ "%s = %s",
+ quote_qualified_identifier(NULL, count_col),
+ get_operation_string(IVM_SUB, count_col, "mv", "t", NULL, NULL)
+ );
+
+ /* sum = mv.sum + diff.sum */
+ appendStringInfo(&update_aggs_new,
+ "%s = %s, ",
+ quote_qualified_identifier(NULL, resname),
+ get_operation_string(IVM_ADD, resname, "mv", "diff", count_col, NULL)
+ );
+ /* count = mv.count + diff.count */
+ appendStringInfo(&update_aggs_new,
+ "%s = %s",
+ quote_qualified_identifier(NULL, count_col),
+ get_operation_string(IVM_ADD, count_col, "mv", "diff", NULL, NULL)
+ );
+
+ appendStringInfo(&diff_aggs_buf, "%s, %s",
+ quote_qualified_identifier("diff", resname),
+ quote_qualified_identifier("diff", makeObjectName("__ivm_count",resname,"_"))
+ );
+ }
+ else if (!strcmp(aggname, "avg"))
+ {
+ char *sum_col = IVM_colname("sum", resname);
+ char *count_col = IVM_colname("count", resname);
+
+ /* avg = (mv.sum - t.sum)::aggtype / (mv.count - t.count) */
+ appendStringInfo(&update_aggs_old,
+ "%s = %s / %s, ",
+ quote_qualified_identifier(NULL, resname),
+ get_operation_string(IVM_SUB, sum_col, "mv", "t", count_col, aggtype),
+ get_operation_string(IVM_SUB, count_col, "mv", "t", NULL, NULL)
+ );
+ /* sum = mv.sum - t.sum */
+ appendStringInfo(&update_aggs_old,
+ "%s = %s, ",
+ quote_qualified_identifier(NULL, sum_col),
+ get_operation_string(IVM_SUB, sum_col, "mv", "t", count_col, NULL)
+ );
+ /* count = mv.count - t.count */
+ appendStringInfo(&update_aggs_old,
+ "%s = %s",
+ quote_qualified_identifier(NULL, count_col),
+ get_operation_string(IVM_SUB, count_col, "mv", "t", NULL, NULL)
+ );
+
+ /* avg = (mv.sum + diff.sum)::aggtype / (mv.count + diff.count) */
+ appendStringInfo(&update_aggs_new,
+ "%s = %s / %s, ",
+ quote_qualified_identifier(NULL, resname),
+ get_operation_string(IVM_ADD, sum_col, "mv", "diff", count_col, aggtype),
+ get_operation_string(IVM_ADD, count_col, "mv", "diff", NULL, NULL)
+ );
+ /* sum = mv.sum + diff.sum */
+ appendStringInfo(&update_aggs_new,
+ "%s = %s, ",
+ quote_qualified_identifier(NULL, sum_col),
+ get_operation_string(IVM_ADD, sum_col, "mv", "diff", count_col, NULL)
+ );
+ /* count = mv.count + diff.count */
+ appendStringInfo(&update_aggs_new,
+ "%s = %s",
+ quote_qualified_identifier(NULL, count_col),
+ get_operation_string(IVM_ADD, count_col, "mv", "diff", NULL, NULL)
+ );
+
+ appendStringInfo(&diff_aggs_buf, "%s, %s, %s",
+ quote_qualified_identifier("diff", resname),
+ quote_qualified_identifier("diff", makeObjectName("__ivm_sum",resname,"_")),
+ quote_qualified_identifier("diff", makeObjectName("__ivm_count",resname,"_"))
+ );
+ }
+ else if (!strcmp(aggname, "min") || !strcmp(aggname, "max"))
+ {
+ bool is_min = !strcmp(aggname, "min");
+ char *count_col = IVM_colname("count", resname);
+
+ if (!has_min_or_max)
+ {
+ has_min_or_max = true;
+ appendStringInfo(&returning_buf, "RETURNING mv.ctid, (");
+ }
+ else
+ {
+ appendStringInfo(&returning_buf, " OR ");
+ appendStringInfo(&min_or_max_buf, ",");
+ }
+
+ appendStringInfo(&returning_buf, "%s %s %s",
+ quote_qualified_identifier("mv", resname),
+ is_min ? ">=" : "<=",
+ quote_qualified_identifier("t", resname)
+ );
+ appendStringInfo(&min_or_max_buf, "%s", quote_qualified_identifier(NULL, resname));
+
+ /* Even if the new values is not NULL, this might be recomputated afterwords. */
+ appendStringInfo(&update_aggs_old,
+ "%s = CASE WHEN %s THEN NULL ELSE %s END, ",
+ quote_qualified_identifier(NULL, resname),
+ get_null_condition_string(IVM_SUB, "mv", "t", count_col),
+ quote_qualified_identifier("mv", resname)
+ );
+ /* count = mv.count - t.count */
+ appendStringInfo(&update_aggs_old,
+ "%s = %s",
+ quote_qualified_identifier(NULL, count_col),
+ get_operation_string(IVM_SUB, count_col, "mv", "t", NULL, NULL)
+ );
+
+ /*
+ * min = least(mv.min, diff.min)
+ * max = greatest(mv.max, diff.max)
+ */
+ appendStringInfo(&update_aggs_new,
+ "%s = CASE WHEN %s THEN NULL ELSE %s(%s,%s) END, ",
+ quote_qualified_identifier(NULL, resname),
+ get_null_condition_string(IVM_ADD, "mv", "diff", count_col),
+
+ is_min ? "least" : "greatest",
+ quote_qualified_identifier("mv", resname),
+ quote_qualified_identifier("diff", resname)
+ );
+
+ /* count = mv.count + diff.count */
+ appendStringInfo(&update_aggs_new,
+ "%s = %s",
+ quote_qualified_identifier(NULL, count_col),
+ get_operation_string(IVM_ADD, count_col, "mv", "diff", NULL, NULL)
+ );
+
+ appendStringInfo(&diff_aggs_buf, "%s, %s",
+ quote_qualified_identifier("diff", resname),
+ quote_qualified_identifier("diff", makeObjectName("__ivm_count",resname,"_"))
+ );
+
+ num_min_or_max++;
+ }
+ else
+ elog(ERROR, "unsupported aggregate function: %s", aggname);
+
+ }
+ }
+ if (has_min_or_max)
+ appendStringInfo(&returning_buf, ") AS recalc");
+
+ if (query->hasAggs)
+ {
+ initStringInfo(&mv_gkeys_buf);
+ initStringInfo(&diff_gkeys_buf);
+ initStringInfo(&updt_gkeys_buf);
+
+ if (with_group)
+ {
+ sep_agg= "";
+ foreach (lc, query->groupClause)
+ {
+ SortGroupClause *sgcl = (SortGroupClause *) lfirst(lc);
+ TargetEntry *tle = get_sortgroupclause_tle(sgcl, query->targetList);
+
+ Form_pg_attribute attr = TupleDescAttr(matviewRel->rd_att, tle->resno-1);
+ char *resname = NameStr(attr->attname);
+
+ appendStringInfo(&mv_gkeys_buf, "%s", sep_agg);
+ appendStringInfo(&diff_gkeys_buf, "%s", sep_agg);
+ appendStringInfo(&updt_gkeys_buf, "%s", sep_agg);
+
+ sep_agg = ", ";
+
+ appendStringInfo(&mv_gkeys_buf, "%s", quote_qualified_identifier("mv", resname));
+ appendStringInfo(&diff_gkeys_buf, "%s", quote_qualified_identifier("diff", resname));
+ appendStringInfo(&updt_gkeys_buf, "%s", quote_qualified_identifier("updt", resname));
+
+ num_group_keys++;
+ }
+
+ if (has_min_or_max)
+ {
+ appendStringInfo(&returning_buf, ", %s", mv_gkeys_buf.data);
+ appendStringInfo(&result_buf, "SELECT ctid AS tid, %s FROM updt WHERE recalc", updt_gkeys_buf.data);
+ }
+ }
+ else
+ {
+ appendStringInfo(&mv_gkeys_buf, "1");
+ appendStringInfo(&diff_gkeys_buf, "1");
+ appendStringInfo(&updt_gkeys_buf, "1");
+ }
+ }
+
+
+ /* Open SPI context. */
+ if (SPI_connect() != SPI_OK_CONNECT)
+ elog(ERROR, "SPI_connect failed");
+
+ /* Analyze the temp table with the new contents. */
+ if (tempname_new)
+ {
+ appendStringInfo(&querybuf, "ANALYZE %s", tempname_new);
+ if (SPI_exec(querybuf.data, 0) != SPI_OK_UTILITY)
+ elog(ERROR, "SPI_exec failed: %s", querybuf.data);
+ }
+ if (tempname_old)
+ {
+ resetStringInfo(&querybuf);
+ appendStringInfo(&querybuf, "ANALYZE %s", tempname_old);
+ if (SPI_exec(querybuf.data, 0) != SPI_OK_UTILITY)
+ elog(ERROR, "SPI_exec failed: %s", querybuf.data);
+ }
+
+ OpenMatViewIncrementalMaintenance();
+
+ if (query->hasAggs)
+ {
+ if (tempname_old)
+ {
+ resetStringInfo(&querybuf);
+ appendStringInfo(&querybuf,
+ "WITH t AS ("
+ " SELECT diff.__ivm_count__, "
+ " %s,"
+ " (diff.__ivm_count__ = mv.__ivm_count__) AS for_dlt, "
+ " mv.ctid"
+ " FROM %s AS mv, %s AS diff WHERE (%s) = (%s)"
+ "), updt AS ("
+ " UPDATE %s AS mv SET __ivm_count__ = mv.__ivm_count__ - t.__ivm_count__,"
+ " %s"
+ " FROM t WHERE mv.ctid = t.ctid AND NOT for_dlt"
+ " %s"
+ "), dlt AS ("
+ " DELETE FROM %s AS mv USING t WHERE mv.ctid = t.ctid AND for_dlt "
+ ") %s",
+ diff_aggs_buf.data,
+ matviewname, tempname_old, mv_gkeys_buf.data, diff_gkeys_buf.data,
+ matviewname, update_aggs_old.data,
+ returning_buf.data,
+ matviewname,
+ (has_min_or_max && with_group) ? result_buf.data : "SELECT 1"
+ );
+ if (SPI_exec(querybuf.data, 0) != SPI_OK_SELECT)
+ elog(ERROR, "SPI_exec failed: %s", querybuf.data);
+
+ if (has_min_or_max && SPI_processed > 0)
+ {
+ SPITupleTable *tuptable_recalc = SPI_tuptable;
+ TupleDesc tupdesc_recalc = tuptable_recalc->tupdesc;
+ int64 processed = SPI_processed;
+ uint64 i;
+ Oid *keyTypes = NULL, *minmaxTypes = NULL;
+ char *keyNulls = NULL, *minmaxNulls = NULL;
+ Datum *keyVals = NULL, *minmaxVals = NULL;
+
+ if (with_group)
+ {
+ keyTypes = palloc(sizeof(Oid) * num_group_keys);
+ keyNulls = palloc(sizeof(char) * num_group_keys);
+ keyVals = palloc(sizeof(Datum) * num_group_keys);
+ Assert(tupdesc_recalc->natts == num_group_keys + 1);
+
+ for (i = 0; i < num_group_keys; i++)
+ keyTypes[i] = TupleDescAttr(tupdesc_recalc, i+1)->atttypid;
+ }
+
+ minmaxTypes = palloc(sizeof(Oid) * (num_min_or_max + 1));
+ minmaxNulls = palloc(sizeof(char) * (num_min_or_max + 1));
+ minmaxVals = palloc(sizeof(Datum) * (num_min_or_max + 1));
+
+ for (i=0; i< processed; i++)
+ {
+ int j;
+ bool isnull;
+ SPIPlanPtr plan;
+ SPITupleTable *tuptable_minmax;
+ TupleDesc tupdesc_minmax;
+
+ if (with_group)
+ {
+ for (j = 0; j < num_group_keys; j++)
+ {
+ keyVals[j] = SPI_getbinval(tuptable_recalc->vals[i], tupdesc_recalc, j+2, &isnull);
+ if (isnull)
+ keyNulls[j] = 'n';
+ else
+ keyNulls[j] = ' ';
+ }
+ }
+
+ plan = get_plan_for_recalc_min_max(matviewOid, min_or_max_buf.data,
+ mv_gkeys_buf.data, num_group_keys, keyTypes, with_group);
+
+ if (SPI_execute_plan(plan, keyVals, keyNulls, false, 0) != SPI_OK_SELECT)
+ elog(ERROR, "SPI_execcute_plan1");
+ if (SPI_processed != 1)
+ elog(ERROR, "SPI_execcute_plan returned zere or more than one rows");
+
+ tuptable_minmax = SPI_tuptable;
+ tupdesc_minmax = tuptable_minmax->tupdesc;
+
+ Assert(tupdesc_minmax->natts == num_min_or_max);
+
+ for (j = 0; j < tupdesc_minmax->natts; j++)
+ {
+ if (i == 0)
+ minmaxTypes[j] = TupleDescAttr(tupdesc_minmax, j)->atttypid;
+
+ minmaxVals[j] = SPI_getbinval(tuptable_minmax->vals[0], tupdesc_minmax, j+1, &isnull);
+ if (isnull)
+ minmaxNulls[j] = 'n';
+ else
+ minmaxNulls[j] = ' ';
+ }
+ minmaxTypes[j] = TIDOID;
+ minmaxVals[j] = SPI_getbinval(tuptable_recalc->vals[i], tupdesc_recalc, 1, &isnull);
+ minmaxNulls[j] = ' ';
+
+ plan = get_plan_for_set_min_max(matviewOid, matviewname, min_or_max_buf.data,
+ num_min_or_max, minmaxTypes, with_group);
+
+ if (SPI_execute_plan(plan, minmaxVals, minmaxNulls, false, 0) != SPI_OK_UPDATE)
+ elog(ERROR, "SPI_execcute_plan2");
+
+ }
+ }
+
+ }
+ if (tempname_new)
+ {
+ resetStringInfo(&querybuf);
+ appendStringInfo(&querybuf,
+ "WITH updt AS ("
+ " UPDATE %s AS mv SET __ivm_count__ = mv.__ivm_count__ + diff.__ivm_count__"
+ ", %s "
+ " FROM %s AS diff WHERE (%s) = (%s)"
+ " RETURNING %s"
+ ") INSERT INTO %s (SELECT * FROM %s AS diff WHERE (%s) NOT IN (SELECT %s FROM updt));",
+ matviewname, update_aggs_new.data, tempname_new,
+ mv_gkeys_buf.data, diff_gkeys_buf.data, diff_gkeys_buf.data,
+ matviewname, tempname_new, diff_gkeys_buf.data, updt_gkeys_buf.data);
+ if (SPI_exec(querybuf.data, 0) != SPI_OK_INSERT)
+ elog(ERROR, "SPI_exec failed: %s", querybuf.data);
+ }
+ }
+ else
+ {
+ if (tempname_old)
+ {
+ resetStringInfo(&querybuf);
+ appendStringInfo(&querybuf,
+ "WITH t AS ("
+ " SELECT diff.__ivm_count__, (diff.__ivm_count__ = mv.__ivm_count__) AS for_dlt, mv.ctid"
+ " FROM %s AS mv, %s AS diff WHERE (%s) = (%s)"
+ "), updt AS ("
+ " UPDATE %s AS mv SET __ivm_count__ = mv.__ivm_count__ - t.__ivm_count__"
+ " FROM t WHERE mv.ctid = t.ctid AND NOT for_dlt"
+ ") DELETE FROM %s AS mv USING t WHERE mv.ctid = t.ctid AND for_dlt;",
+ matviewname, tempname_old, mvatts_buf.data, diffatts_buf.data, matviewname, matviewname);
+ if (SPI_exec(querybuf.data, 0) != SPI_OK_DELETE)
+ elog(ERROR, "SPI_exec failed: %s", querybuf.data);
+ }
+ if (tempname_new)
+ {
+ resetStringInfo(&querybuf);
+ appendStringInfo(&querybuf,
+ "WITH updt AS ("
+ " UPDATE %s AS mv SET __ivm_count__ = mv.__ivm_count__ + diff.__ivm_count__"
+ " FROM %s AS diff WHERE (%s) = (%s)"
+ " RETURNING %s"
+ ") INSERT INTO %s (SELECT * FROM %s AS diff WHERE (%s) NOT IN (SELECT * FROM updt));",
+ matviewname, tempname_new, mvatts_buf.data, diffatts_buf.data, diffatts_buf.data, matviewname, tempname_new, diffatts_buf.data);
+ if (SPI_exec(querybuf.data, 0) != SPI_OK_INSERT)
+ elog(ERROR, "SPI_exec failed: %s", querybuf.data);
+ }
+ }
+
+ /* We're done maintaining the materialized view. */
+ CloseMatViewIncrementalMaintenance();
+
+ if (OidIsValid(tempOid_new))
+ table_close(tempRel_new, NoLock);
+ if (OidIsValid(tempOid_old))
+ table_close(tempRel_old, NoLock);
+
+ table_close(matviewRel, NoLock);
+
+
+ /* Close SPI context. */
+ if (SPI_finish() != SPI_OK_FINISH)
+ elog(ERROR, "SPI_finish failed");
+}
+
+
+static void
+mv_InitHashTables(void)
+{
+ HASHCTL ctl;
+
+ memset(&ctl, 0, sizeof(ctl));
+ ctl.keysize = sizeof(MV_QueryKey);
+ ctl.entrysize = sizeof(MV_QueryHashEntry);
+ mv_query_cache = hash_create("MV query cache",
+ MV_INIT_QUERYHASHSIZE,
+ &ctl, HASH_ELEM | HASH_BLOBS);
+
+ memset(&ctl, 0, sizeof(ctl));
+ ctl.keysize = sizeof(Oid);
+ ctl.entrysize = sizeof(MV_TriggerHashEntry);
+ mv_trigger_info = hash_create("MV trigger info",
+ MV_INIT_QUERYHASHSIZE,
+ &ctl, HASH_ELEM | HASH_BLOBS);
+}
+
+static SPIPlanPtr
+mv_FetchPreparedPlan(MV_QueryKey *key)
+{
+ MV_QueryHashEntry *entry;
+ SPIPlanPtr plan;
+
+ /*
+ * On the first call initialize the hashtable
+ */
+ if (!mv_query_cache)
+ mv_InitHashTables();
+
+ /*
+ * Lookup for the key
+ */
+ entry = (MV_QueryHashEntry *) hash_search(mv_query_cache,
+ (void *) key,
+ HASH_FIND, NULL);
+ if (entry == NULL)
+ return NULL;
+
+ /*
+ * Check whether the plan is still valid. If it isn't, we don't want to
+ * simply rely on plancache.c to regenerate it; rather we should start
+ * from scratch and rebuild the query text too. This is to cover cases
+ * such as table/column renames. We depend on the plancache machinery to
+ * detect possible invalidations, though.
+ *
+ * CAUTION: this check is only trustworthy if the caller has already
+ * locked both materialized views and base tables.
+ */
+ plan = entry->plan;
+ if (plan && SPI_plan_is_valid(plan))
+ return plan;
+
+ /*
+ * Otherwise we might as well flush the cached plan now, to free a little
+ * memory space before we make a new one.
+ */
+ entry->plan = NULL;
+ if (plan)
+ SPI_freeplan(plan);
+
+ return NULL;
+}
+
+/*
+ * mv_HashPreparedPlan -
+ *
+ * Add another plan to our private SPI query plan hashtable.
+ */
+static void
+mv_HashPreparedPlan(MV_QueryKey *key, SPIPlanPtr plan)
+{
+ MV_QueryHashEntry *entry;
+ bool found;
+
+ /*
+ * On the first call initialize the hashtable
+ */
+ if (!mv_query_cache)
+ mv_InitHashTables();
+
+ /*
+ * Add the new plan. We might be overwriting an entry previously found
+ * invalid by mv_FetchPreparedPlan.
+ */
+ entry = (MV_QueryHashEntry *) hash_search(mv_query_cache,
+ (void *) key,
+ HASH_ENTER, &found);
+ Assert(!found || entry->plan == NULL);
+ entry->plan = plan;
+}
+
+/* ----------
+ * mv_BuildQueryKey -
+ *
+ * Construct a hashtable key for a prepared SPI plan for IVM.
+ *
+ * key: output argument, *key is filled in based on the other arguments
+ * matview_id: OID of materialized view
+ * query_type: an internal number identifying the query type
+ * (see MV_PLAN_XXX constants at head of file)
+ * ----------
+ */
+static void
+mv_BuildQueryKey(MV_QueryKey *key, Oid matview_id, int32 query_type)
+{
+ /*
+ * We assume struct MV_QueryKey contains no padding bytes, else we'd need
+ * to use memset to clear them.
+ */
+ key->matview_id = matview_id;
+ key->query_type = query_type;
+}
+
+static SPIPlanPtr
+get_plan_for_recalc_min_max(Oid matviewOid, const char *min_max_list,
+ const char *group_keys, int nkeys, Oid *keyTypes, bool with_group)
+{
+ MV_QueryKey key;
+ SPIPlanPtr plan;
+
+ /* Fetch or prepare a saved plan for the real check */
+ mv_BuildQueryKey(&key, matviewOid, MV_PLAN_RECALC_MINMAX);
+
+ if ((plan = mv_FetchPreparedPlan(&key)) == NULL)
+ {
+ char *viewdef;
+ StringInfoData str;
+ int i;
+
+
+ /* get view definition of matview */
+ viewdef = text_to_cstring((text *) DatumGetPointer(
+ DirectFunctionCall1(pg_get_viewdef, ObjectIdGetDatum(matviewOid))));
+ /* get rid of tailling semi-collon */
+ viewdef[strlen(viewdef)-1] = '\0';
+
+ initStringInfo(&str);
+ appendStringInfo(&str, "SELECT %s FROM (%s) mv", min_max_list, viewdef);
+
+ if (with_group)
+ {
+ appendStringInfo(&str, " WHERE (%s) = (", group_keys);
+
+ for (i = 1; i <= nkeys; i++)
+ appendStringInfo(&str, "%s$%d", (i==1 ? "" : ", "),i );
+
+ appendStringInfo(&str, ")");
+ }
+
+ plan = SPI_prepare(str.data, (with_group ? nkeys : 0), (with_group ? keyTypes : NULL));
+ if (plan == NULL)
+ elog(ERROR, "SPI_prepare returned %s for %s", SPI_result_code_string(SPI_result), str.data);
+
+ SPI_keepplan(plan);
+ mv_HashPreparedPlan(&key, plan);
+ }
+
+ return plan;
+}
+
+static SPIPlanPtr
+get_plan_for_set_min_max(Oid matviewOid, char *matviewname, const char *min_max_list,
+ int num_min_max, Oid *valTypes, bool with_group)
+{
+ MV_QueryKey key;
+ SPIPlanPtr plan;
+
+ /* Fetch or prepare a saved plan for the real check */
+ mv_BuildQueryKey(&key, matviewOid, MV_PLAN_SET_MINMAX);
+
+ if ((plan = mv_FetchPreparedPlan(&key)) == NULL)
+ {
+ StringInfoData str;
+ int i;
+
+ initStringInfo(&str);
+ appendStringInfo(&str, "UPDATE %s AS mv SET (%s) = (",
+ matviewname, min_max_list);
+
+ for (i = 1; i <= num_min_max; i++)
+ appendStringInfo(&str, "%s$%d", (i==1 ? "" : ", "), i);
+
+ appendStringInfo(&str, ")");
+
+ if (with_group)
+ appendStringInfo(&str, " WHERE ctid = $%d", num_min_max + 1);
+
+ plan = SPI_prepare(str.data, num_min_max + (with_group ? 1 : 0), valTypes);
+ if (plan == NULL)
+ elog(ERROR, "SPI_prepare returned %s for %s", SPI_result_code_string(SPI_result), str.data);
+
+ SPI_keepplan(plan);
+ mv_HashPreparedPlan(&key, plan);
+ }
+
+ return plan;
+}
+
+
+/*
+ * get_matview_query - get the Query from a matview's _RETURN rule.
+ */
+static Query *
+get_matview_query(Relation matviewRel)
+{
+ RewriteRule *rule;
+ List * actions;
+
+ /*
+ * Check that everything is correct for a refresh. Problems at this point
+ * are internal errors, so elog is sufficient.
+ */
+ if (matviewRel->rd_rel->relhasrules == false ||
+ matviewRel->rd_rules->numLocks < 1)
+ elog(ERROR,
+ "materialized view \"%s\" is missing rewrite information",
+ RelationGetRelationName(matviewRel));
+
+ if (matviewRel->rd_rules->numLocks > 1)
+ elog(ERROR,
+ "materialized view \"%s\" has too many rules",
+ RelationGetRelationName(matviewRel));
+
+ rule = matviewRel->rd_rules->rules[0];
+ if (rule->event != CMD_SELECT || !(rule->isInstead))
+ elog(ERROR,
+ "the rule for materialized view \"%s\" is not a SELECT INSTEAD OF rule",
+ RelationGetRelationName(matviewRel));
+
+ actions = rule->actions;
+ if (list_length(actions) != 1)
+ elog(ERROR,
+ "the rule for materialized view \"%s\" is not a single action",
+ RelationGetRelationName(matviewRel));
+
+ /*
+ * The stored query was rewritten at the time of the MV definition, but
+ * has not been scribbled on by the planner.
+ */
+ return linitial_node(Query, actions);
+}
+
+static void
+truncate_view_delta(Oid delta_oid)
+{
+ Relation rel;
+
+ if (!OidIsValid(delta_oid))
+ return;
+
+ rel = table_open(delta_oid, NoLock);
+ ExecuteTruncateGuts(list_make1(rel), list_make1_oid(delta_oid), NIL,
+ DROP_RESTRICT, false);
+ table_close(rel, NoLock);
+}
+
+void
+AtAbort_IVM()
+{
+ HASH_SEQ_STATUS seq;
+ MV_TriggerHashEntry *entry;
+
+ if (mv_trigger_info)
+ {
+ hash_seq_init(&seq, mv_trigger_info);
+ while ((entry = hash_seq_search(&seq)) != NULL)
+ clean_up_IVM_hash_entry(entry);
+ }
+}
+
+static void
+clean_up_IVM_hash_entry(MV_TriggerHashEntry *entry)
+{
+ bool found;
+ ListCell *lc;
+
+ foreach(lc, entry->tables)
+ {
+ MV_TriggerTable *table = (MV_TriggerTable *) lfirst(lc);
+
+ list_free(table->old_tuplestores);
+ list_free(table->new_tuplestores);
+ }
+ list_free(entry->tables);
+
+ hash_search(mv_trigger_info, (void *) &entry->matview_id, HASH_REMOVE, &found);
+}
+
+static void
+clean_up_IVM_temptable(Oid tempOid_old, Oid tempOid_new)
+{
+ StringInfoData querybuf;
+ Relation tempRel_old, tempRel_new;
+ char *tempname_old = NULL, *tempname_new = NULL;
+
+ if (OidIsValid(tempOid_new))
+ {
+ tempRel_new = table_open(tempOid_new, NoLock);
+ tempname_new = quote_qualified_identifier(get_namespace_name(RelationGetNamespace(tempRel_new)),
+ RelationGetRelationName(tempRel_new));
+ table_close(tempRel_new, NoLock);
+ }
+ if (OidIsValid(tempOid_old))
+ {
+ tempRel_old = table_open(tempOid_old, NoLock);
+ tempname_old = quote_qualified_identifier(get_namespace_name(RelationGetNamespace(tempRel_old)),
+ RelationGetRelationName(tempRel_old));
+ table_close(tempRel_old, NoLock);
+ }
+
+ initStringInfo(&querybuf);
+
+ /* Open SPI context. */
+ if (SPI_connect() != SPI_OK_CONNECT)
+ elog(ERROR, "SPI_connect failed");
+
+ /* Clean up temp tables. */
+ if (OidIsValid(tempOid_old))
+ {
+ resetStringInfo(&querybuf);
+ appendStringInfo(&querybuf, "DROP TABLE %s", tempname_old);
+ if (SPI_exec(querybuf.data, 0) != SPI_OK_UTILITY)
+ elog(ERROR, "SPI_exec failed: %s", querybuf.data);
+ }
+ if (OidIsValid(tempOid_new))
+ {
+ resetStringInfo(&querybuf);
+ appendStringInfo(&querybuf, "DROP TABLE %s", tempname_new);
+ if (SPI_exec(querybuf.data, 0) != SPI_OK_UTILITY)
+ elog(ERROR, "SPI_exec failed: %s", querybuf.data);
+ }
+
+ /* Close SPI context. */
+ if (SPI_finish() != SPI_OK_FINISH)
+ elog(ERROR, "SPI_finish failed");
+}
diff --git a/src/backend/commands/trigger.c b/src/backend/commands/trigger.c
index cdb1105b4a..096ab1b67c 100644
--- a/src/backend/commands/trigger.c
+++ b/src/backend/commands/trigger.c
@@ -3803,6 +3803,7 @@ typedef struct AfterTriggersData
AfterTriggersQueryData *query_stack; /* array of structs shown below */
int query_depth; /* current index in above array */
int maxquerydepth; /* allocated len of above array */
+ List *prolonged_tuplestores;
/* per-subtransaction-level data: */
AfterTriggersTransData *trans_stack; /* array of structs shown below */
@@ -3833,6 +3834,7 @@ struct AfterTriggersTableData
bool closed; /* true when no longer OK to add tuples */
bool before_trig_done; /* did we already queue BS triggers? */
bool after_trig_done; /* did we already queue AS triggers? */
+ bool prolonged;
AfterTriggerEventList after_trig_events; /* if so, saved list pointer */
Tuplestorestate *old_tuplestore; /* "old" transition table, if any */
Tuplestorestate *new_tuplestore; /* "new" transition table, if any */
@@ -4586,6 +4588,35 @@ afterTriggerInvokeEvents(AfterTriggerEventList *events,
return all_fired;
}
+void
+SetTransitionTablePreserved(Oid relid, CmdType cmdType)
+{
+ AfterTriggersTableData *table;
+ AfterTriggersQueryData *qs;
+ bool found = false;
+ ListCell *lc;
+
+ /* Check state, like AfterTriggerSaveEvent. */
+ if (afterTriggers.query_depth < 0)
+ elog(ERROR, "SetTransitionTablePreserved() called outside of query");
+
+ qs = &afterTriggers.query_stack[afterTriggers.query_depth];
+
+ foreach(lc, qs->tables)
+ {
+ table = (AfterTriggersTableData *) lfirst(lc);
+ if (table->relid == relid && table->cmdType == cmdType &&
+ table->closed)
+ {
+ table->prolonged = true;
+ found = true;
+ }
+
+ }
+
+ if (!found)
+ elog(ERROR,"could not find table with OID %d and command type %d", relid, cmdType);
+}
/*
* GetAfterTriggersTableData
@@ -4755,6 +4786,7 @@ AfterTriggerBeginXact(void)
*/
afterTriggers.firing_counter = (CommandId) 1; /* mustn't be 0 */
afterTriggers.query_depth = -1;
+ afterTriggers.prolonged_tuplestores = NIL;
/*
* Verify that there is no leftover state remaining. If these assertions
@@ -4915,11 +4947,29 @@ AfterTriggerFreeQuery(AfterTriggersQueryData *qs)
ts = table->old_tuplestore;
table->old_tuplestore = NULL;
if (ts)
- tuplestore_end(ts);
+ {
+ if (table->prolonged && afterTriggers.query_depth > 0)
+ {
+ MemoryContext oldcxt = MemoryContextSwitchTo(CurTransactionContext);
+ afterTriggers.prolonged_tuplestores = lappend(afterTriggers.prolonged_tuplestores, ts);
+ MemoryContextSwitchTo(oldcxt);
+ }
+ else
+ tuplestore_end(ts);
+ }
ts = table->new_tuplestore;
table->new_tuplestore = NULL;
if (ts)
- tuplestore_end(ts);
+ {
+ if (table->prolonged && afterTriggers.query_depth > 0)
+ {
+ MemoryContext oldcxt = MemoryContextSwitchTo(CurTransactionContext);
+ afterTriggers.prolonged_tuplestores = lappend(afterTriggers.prolonged_tuplestores, ts);
+ MemoryContextSwitchTo(oldcxt);
+ }
+ else
+ tuplestore_end(ts);
+ }
}
/*
@@ -4929,6 +4979,17 @@ AfterTriggerFreeQuery(AfterTriggersQueryData *qs)
*/
qs->tables = NIL;
list_free_deep(tables);
+
+ if (afterTriggers.query_depth == 0)
+ {
+ foreach(lc, afterTriggers.prolonged_tuplestores)
+ {
+ ts = (Tuplestorestate *) lfirst(lc);
+ if (ts)
+ tuplestore_end(ts);
+ }
+ afterTriggers.prolonged_tuplestores = NIL;
+ }
}
@@ -6024,6 +6085,9 @@ AfterTriggerSaveEvent(EState *estate, ResultRelInfo *relinfo,
else
new_shared.ats_table = NULL;
+ if (new_shared.ats_table != NULL && trigger->tgfoid == F_IVM_IMMEDIATE_MAINTENANCE)
+ new_shared.ats_table->prolonged = true;
+
afterTriggerAddEvent(&afterTriggers.query_stack[afterTriggers.query_depth].events,
&new_event, &new_shared);
}
diff --git a/src/backend/nodes/copyfuncs.c b/src/backend/nodes/copyfuncs.c
index 3432bb921d..a872b59f13 100644
--- a/src/backend/nodes/copyfuncs.c
+++ b/src/backend/nodes/copyfuncs.c
@@ -2365,6 +2365,7 @@ _copyRangeTblEntry(const RangeTblEntry *from)
COPY_SCALAR_FIELD(relkind);
COPY_SCALAR_FIELD(rellockmode);
COPY_NODE_FIELD(tablesample);
+ COPY_SCALAR_FIELD(relisivm);
COPY_NODE_FIELD(subquery);
COPY_SCALAR_FIELD(security_barrier);
COPY_SCALAR_FIELD(jointype);
diff --git a/src/backend/nodes/equalfuncs.c b/src/backend/nodes/equalfuncs.c
index 18cb014373..503915b9ef 100644
--- a/src/backend/nodes/equalfuncs.c
+++ b/src/backend/nodes/equalfuncs.c
@@ -2651,6 +2651,7 @@ _equalRangeTblEntry(const RangeTblEntry *a, const RangeTblEntry *b)
COMPARE_SCALAR_FIELD(relkind);
COMPARE_SCALAR_FIELD(rellockmode);
COMPARE_NODE_FIELD(tablesample);
+ COMPARE_SCALAR_FIELD(relisivm);
COMPARE_NODE_FIELD(subquery);
COMPARE_SCALAR_FIELD(security_barrier);
COMPARE_SCALAR_FIELD(jointype);
diff --git a/src/backend/nodes/outfuncs.c b/src/backend/nodes/outfuncs.c
index b0dcd02ff6..52e898d615 100644
--- a/src/backend/nodes/outfuncs.c
+++ b/src/backend/nodes/outfuncs.c
@@ -3058,6 +3058,7 @@ _outRangeTblEntry(StringInfo str, const RangeTblEntry *node)
WRITE_CHAR_FIELD(relkind);
WRITE_INT_FIELD(rellockmode);
WRITE_NODE_FIELD(tablesample);
+ WRITE_BOOL_FIELD(relisivm);
break;
case RTE_SUBQUERY:
WRITE_NODE_FIELD(subquery);
diff --git a/src/backend/nodes/readfuncs.c b/src/backend/nodes/readfuncs.c
index 764e3bb90c..3301ccfbc8 100644
--- a/src/backend/nodes/readfuncs.c
+++ b/src/backend/nodes/readfuncs.c
@@ -1366,6 +1366,7 @@ _readRangeTblEntry(void)
READ_CHAR_FIELD(relkind);
READ_INT_FIELD(rellockmode);
READ_NODE_FIELD(tablesample);
+ READ_BOOL_FIELD(relisivm);
break;
case RTE_SUBQUERY:
READ_NODE_FIELD(subquery);
diff --git a/src/backend/parser/gram.y b/src/backend/parser/gram.y
index 3f67aaf30e..2ccdda580a 100644
--- a/src/backend/parser/gram.y
+++ b/src/backend/parser/gram.y
@@ -418,6 +418,7 @@ static Node *makeRecursiveViewSelect(char *relname, List *aliases, Node *query);
%type <range> OptTempTableName
%type <into> into_clause create_as_target create_mv_target
+%type <boolean> incremental
%type <defelt> createfunc_opt_item common_func_opt_item dostmt_opt_item
%type <fun_param> func_arg func_arg_with_default table_func_column aggr_arg
@@ -645,7 +646,7 @@ static Node *makeRecursiveViewSelect(char *relname, List *aliases, Node *query);
HANDLER HAVING HEADER_P HOLD HOUR_P
IDENTITY_P IF_P ILIKE IMMEDIATE IMMUTABLE IMPLICIT_P IMPORT_P IN_P INCLUDE
- INCLUDING INCREMENT INDEX INDEXES INHERIT INHERITS INITIALLY INLINE_P
+ INCLUDING INCREMENT INCREMENTAL INDEX INDEXES INHERIT INHERITS INITIALLY INLINE_P
INNER_P INOUT INPUT_P INSENSITIVE INSERT INSTEAD INT_P INTEGER
INTERSECT INTERVAL INTO INVOKER IS ISNULL ISOLATION
@@ -4083,30 +4084,32 @@ opt_with_data:
*****************************************************************************/
CreateMatViewStmt:
- CREATE OptNoLog MATERIALIZED VIEW create_mv_target AS SelectStmt opt_with_data
+ CREATE OptNoLog incremental MATERIALIZED VIEW create_mv_target AS SelectStmt opt_with_data
{
CreateTableAsStmt *ctas = makeNode(CreateTableAsStmt);
- ctas->query = $7;
- ctas->into = $5;
+ ctas->query = $8;
+ ctas->into = $6;
ctas->relkind = OBJECT_MATVIEW;
ctas->is_select_into = false;
ctas->if_not_exists = false;
/* cram additional flags into the IntoClause */
- $5->rel->relpersistence = $2;
- $5->skipData = !($8);
+ $6->rel->relpersistence = $2;
+ $6->skipData = !($9);
+ $6->ivm = $3;
$$ = (Node *) ctas;
}
- | CREATE OptNoLog MATERIALIZED VIEW IF_P NOT EXISTS create_mv_target AS SelectStmt opt_with_data
+ | CREATE OptNoLog incremental MATERIALIZED VIEW IF_P NOT EXISTS create_mv_target AS SelectStmt opt_with_data
{
CreateTableAsStmt *ctas = makeNode(CreateTableAsStmt);
- ctas->query = $10;
- ctas->into = $8;
+ ctas->query = $11;
+ ctas->into = $9;
ctas->relkind = OBJECT_MATVIEW;
ctas->is_select_into = false;
ctas->if_not_exists = true;
/* cram additional flags into the IntoClause */
- $8->rel->relpersistence = $2;
- $8->skipData = !($11);
+ $9->rel->relpersistence = $2;
+ $9->skipData = !($12);
+ $9->ivm = $3;
$$ = (Node *) ctas;
}
;
@@ -4123,9 +4126,14 @@ create_mv_target:
$$->tableSpaceName = $5;
$$->viewQuery = NULL; /* filled at analysis time */
$$->skipData = false; /* might get changed later */
+ $$->ivm = false;
}
;
+incremental: INCREMENTAL { $$ = true; }
+ | /*EMPTY*/ { $$ = false; }
+ ;
+
OptNoLog: UNLOGGED { $$ = RELPERSISTENCE_UNLOGGED; }
| /*EMPTY*/ { $$ = RELPERSISTENCE_PERMANENT; }
;
@@ -15157,6 +15165,7 @@ unreserved_keyword:
| INCLUDE
| INCLUDING
| INCREMENT
+ | INCREMENTAL
| INDEX
| INDEXES
| INHERIT
diff --git a/src/backend/parser/parse_relation.c b/src/backend/parser/parse_relation.c
index 4dd81507a7..a6c8c3fd4d 100644
--- a/src/backend/parser/parse_relation.c
+++ b/src/backend/parser/parse_relation.c
@@ -37,6 +37,7 @@
#include "utils/syscache.h"
#include "utils/varlena.h"
+#include "commands/matview.h"
#define MAX_FUZZY_DISTANCE 3
@@ -56,9 +57,10 @@ static void expandTupleDesc(TupleDesc tupdesc, Alias *eref,
int count, int offset,
int rtindex, int sublevels_up,
int location, bool include_dropped,
- List **colnames, List **colvars);
+ List **colnames, List **colvars, bool is_ivm);
static int specialAttNum(const char *attname);
static bool isQueryUsingTempRelation_walker(Node *node, void *context);
+static bool isIvmColumn(const char *s);
/*
@@ -1241,6 +1243,7 @@ addRangeTableEntry(ParseState *pstate,
rte->relid = RelationGetRelid(rel);
rte->relkind = rel->rd_rel->relkind;
rte->rellockmode = lockmode;
+ rte->relisivm = rel->rd_rel->relisivm;
/*
* Build the list of effective column names using user-supplied aliases
@@ -1320,6 +1323,7 @@ addRangeTableEntryForRelation(ParseState *pstate,
rte->relid = RelationGetRelid(rel);
rte->relkind = rel->rd_rel->relkind;
rte->rellockmode = lockmode;
+ rte->relisivm = rel->rd_rel->relisivm;
/*
* Build the list of effective column names using user-supplied aliases
@@ -2318,7 +2322,7 @@ expandRTE(RangeTblEntry *rte, int rtindex, int sublevels_up,
expandTupleDesc(tupdesc, rte->eref,
rtfunc->funccolcount, atts_done,
rtindex, sublevels_up, location,
- include_dropped, colnames, colvars);
+ include_dropped, colnames, colvars, false);
}
else if (functypclass == TYPEFUNC_SCALAR)
{
@@ -2570,10 +2574,16 @@ expandRelation(Oid relid, Alias *eref, int rtindex, int sublevels_up,
expandTupleDesc(rel->rd_att, eref, rel->rd_att->natts, 0,
rtindex, sublevels_up,
location, include_dropped,
- colnames, colvars);
+ colnames, colvars, RelationIsIVM(rel));
relation_close(rel, AccessShareLock);
}
+static bool
+isIvmColumn(const char *s)
+{
+ return (strncmp(s, "__ivm_", 6) == 0);
+}
+
/*
* expandTupleDesc -- expandRTE subroutine
*
@@ -2587,7 +2597,7 @@ static void
expandTupleDesc(TupleDesc tupdesc, Alias *eref, int count, int offset,
int rtindex, int sublevels_up,
int location, bool include_dropped,
- List **colnames, List **colvars)
+ List **colnames, List **colvars, bool is_ivm)
{
ListCell *aliascell;
int varattno;
@@ -2600,6 +2610,9 @@ expandTupleDesc(TupleDesc tupdesc, Alias *eref, int count, int offset,
{
Form_pg_attribute attr = TupleDescAttr(tupdesc, varattno);
+ if (is_ivm && isIvmColumn(NameStr(attr->attname)) && !MatViewIncrementalMaintenanceIsEnabled())
+ continue;
+
if (attr->attisdropped)
{
if (include_dropped)
diff --git a/src/backend/rewrite/rewriteDefine.c b/src/backend/rewrite/rewriteDefine.c
index 7df2b6154c..5385a038b1 100644
--- a/src/backend/rewrite/rewriteDefine.c
+++ b/src/backend/rewrite/rewriteDefine.c
@@ -765,7 +765,8 @@ checkRuleResultList(List *targetList, TupleDesc resultDesc, bool isSelect,
attr->atttypmod))));
}
- if (i != resultDesc->natts)
+ /* No check for materialized views since this could have special columns for IVM */
+ if ((!isSelect || requireColumnNameMatch) && i != resultDesc->natts)
ereport(ERROR,
(errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
isSelect ?
diff --git a/src/backend/rewrite/rewriteHandler.c b/src/backend/rewrite/rewriteHandler.c
index 3e38007643..5b71f0c665 100644
--- a/src/backend/rewrite/rewriteHandler.c
+++ b/src/backend/rewrite/rewriteHandler.c
@@ -41,6 +41,8 @@
#include "utils/lsyscache.h"
#include "utils/rel.h"
+#include "parser/parser.h"
+#include "commands/matview.h"
/* We use a list of these to detect recursion in RewriteQuery */
typedef struct rewrite_event
@@ -1597,6 +1599,50 @@ ApplyRetrieveRule(Query *parsetree,
if (rule->qual != NULL)
elog(ERROR, "cannot handle qualified ON SELECT rule");
+ if (RelationIsIVM(relation))
+ {
+ rule_action = copyObject(linitial(rule->actions));
+
+ if (!rule_action->distinctClause && !rule_action->groupClause && !rule_action->hasAggs)
+ {
+ StringInfoData str;
+ RawStmt *raw;
+ Query *sub;
+
+ if (rule_action->hasDistinctOn)
+ elog(ERROR, "DISTINCT ON is not supported in IVM");
+
+ initStringInfo(&str);
+ appendStringInfo(&str, "SELECT mv.*, __ivm_count__ FROM %s mv, generate_series(1, mv.__ivm_count__)",
+ quote_qualified_identifier(get_namespace_name(RelationGetNamespace(relation)),
+ RelationGetRelationName(relation)));
+
+ raw = (RawStmt*)linitial(raw_parser(str.data));
+ sub = transformStmt(make_parsestate(NULL),raw->stmt);
+
+ rte = rt_fetch(rt_index, parsetree->rtable);
+
+ rte->rtekind = RTE_SUBQUERY;
+ rte->subquery = sub;
+ rte->security_barrier = RelationIsSecurityView(relation);
+ /* Clear fields that should not be set in a subquery RTE */
+ rte->relid = InvalidOid;
+ rte->relkind = 0;
+ rte->rellockmode = 0;
+ rte->tablesample = NULL;
+ rte->inh = false; /* must not be set for a subquery */
+
+ rte->requiredPerms = 0; /* no permission check on subquery itself */
+ rte->checkAsUser = InvalidOid;
+ rte->selectedCols = NULL;
+ rte->insertedCols = NULL;
+ rte->updatedCols = NULL;
+ rte->extraUpdatedCols = NULL;
+ }
+
+ return parsetree;
+ }
+
if (rt_index == parsetree->resultRelation)
{
/*
@@ -1906,7 +1952,8 @@ fireRIRrules(Query *parsetree, List *activeRIRs)
* In that case this test would need to be postponed till after we've
* opened the rel, so that we could check its state.
*/
- if (rte->relkind == RELKIND_MATVIEW)
+ if (rte->relkind == RELKIND_MATVIEW &&
+ (!rte->relisivm || MatViewIncrementalMaintenanceIsEnabled() || parsetree->commandType != CMD_SELECT))
continue;
/*
diff --git a/src/backend/utils/cache/lsyscache.c b/src/backend/utils/cache/lsyscache.c
index 27602fa49c..694e01b70f 100644
--- a/src/backend/utils/cache/lsyscache.c
+++ b/src/backend/utils/cache/lsyscache.c
@@ -1844,6 +1844,30 @@ get_rel_relispartition(Oid relid)
return false;
}
+/*
+ * get_rel_relisivm
+ *
+ * Returns the relisivm flag associated with a given relation.
+ */
+bool
+get_rel_relisivm(Oid relid)
+{
+ HeapTuple tp;
+
+ tp = SearchSysCache1(RELOID, ObjectIdGetDatum(relid));
+ if (HeapTupleIsValid(tp))
+ {
+ Form_pg_class reltup = (Form_pg_class) GETSTRUCT(tp);
+ bool result;
+
+ result = reltup->relisivm;
+ ReleaseSysCache(tp);
+ return result;
+ }
+ else
+ return false;
+}
+
/*
* get_rel_tablespace
*
diff --git a/src/backend/utils/cache/relcache.c b/src/backend/utils/cache/relcache.c
index 585dcee5db..db04cc98b9 100644
--- a/src/backend/utils/cache/relcache.c
+++ b/src/backend/utils/cache/relcache.c
@@ -1860,6 +1860,8 @@ formrdesc(const char *relationName, Oid relationReltype,
/* ... and they're always populated, too */
relation->rd_rel->relispopulated = true;
+ /* ... and they're always no ivm, too */
+ relation->rd_rel->relisivm = false;
relation->rd_rel->relreplident = REPLICA_IDENTITY_NOTHING;
relation->rd_rel->relpages = 0;
diff --git a/src/bin/psql/describe.c b/src/bin/psql/describe.c
index d7c0fc0c1e..a767b50f26 100644
--- a/src/bin/psql/describe.c
+++ b/src/bin/psql/describe.c
@@ -1492,6 +1492,7 @@ describeOneTableDetails(const char *schemaname,
char relpersistence;
char relreplident;
char *relam;
+ bool isivm;
} tableinfo;
bool show_column_details = false;
@@ -1512,6 +1513,7 @@ describeOneTableDetails(const char *schemaname,
"false AS relhasoids, c.relispartition, %s, c.reltablespace, "
"CASE WHEN c.reloftype = 0 THEN '' ELSE c.reloftype::pg_catalog.regtype::pg_catalog.text END, "
"c.relpersistence, c.relreplident, am.amname\n"
+ ",c.relisivm\n"
"FROM pg_catalog.pg_class c\n "
"LEFT JOIN pg_catalog.pg_class tc ON (c.reltoastrelid = tc.oid)\n"
"LEFT JOIN pg_catalog.pg_am am ON (c.relam = am.oid)\n"
@@ -1688,6 +1690,10 @@ describeOneTableDetails(const char *schemaname,
(char *) NULL : pg_strdup(PQgetvalue(res, 0, 14));
else
tableinfo.relam = NULL;
+ if (pset.sversion >= 130000)
+ tableinfo.isivm = strcmp(PQgetvalue(res, 0, 15), "t") == 0;
+ else
+ tableinfo.isivm = false;
PQclear(res);
res = NULL;
@@ -3292,6 +3298,12 @@ describeOneTableDetails(const char *schemaname,
printfPQExpBuffer(&buf, _("Access method: %s"), tableinfo.relam);
printTableAddFooter(&cont, buf.data);
}
+
+ /* Incremental view maintance info */
+ if (verbose && tableinfo.relkind == RELKIND_MATVIEW && tableinfo.isivm)
+ {
+ printTableAddFooter(&cont, _("Incremental view maintenance: yes"));
+ }
}
/* reloptions, if verbose */
diff --git a/src/bin/psql/tab-complete.c b/src/bin/psql/tab-complete.c
index cb12668276..474b0d482a 100644
--- a/src/bin/psql/tab-complete.c
+++ b/src/bin/psql/tab-complete.c
@@ -1001,6 +1001,7 @@ static const pgsql_thing_t words_after_create[] = {
{"FOREIGN TABLE", NULL, NULL, NULL},
{"FUNCTION", NULL, NULL, Query_for_list_of_functions},
{"GROUP", Query_for_list_of_roles},
+ {"INCREMENTAL MATERIALIZED VIEW", NULL, NULL, &Query_for_list_of_matviews},
{"INDEX", NULL, NULL, &Query_for_list_of_indexes},
{"LANGUAGE", Query_for_list_of_languages},
{"LARGE OBJECT", NULL, NULL, NULL, THING_NO_CREATE | THING_NO_DROP},
@@ -2499,7 +2500,7 @@ psql_completion(const char *text, int start, int end)
COMPLETE_WITH("SEQUENCE", "TABLE", "VIEW");
/* Complete "CREATE UNLOGGED" with TABLE or MATVIEW */
else if (TailMatches("CREATE", "UNLOGGED"))
- COMPLETE_WITH("TABLE", "MATERIALIZED VIEW");
+ COMPLETE_WITH("TABLE", "MATERIALIZED VIEW", "INCREMENTAL MATERIALIZED VIEW");
/* Complete PARTITION BY with RANGE ( or LIST ( or ... */
else if (TailMatches("PARTITION", "BY"))
COMPLETE_WITH("RANGE (", "LIST (", "HASH (");
@@ -2749,13 +2750,16 @@ psql_completion(const char *text, int start, int end)
COMPLETE_WITH("SELECT");
/* CREATE MATERIALIZED VIEW */
- else if (Matches("CREATE", "MATERIALIZED"))
+ else if (Matches("CREATE", "MATERIALIZED") ||
+ Matches("CREATE", "INCREMENTAL", "MATERIALIZED"))
COMPLETE_WITH("VIEW");
- /* Complete CREATE MATERIALIZED VIEW <name> with AS */
- else if (Matches("CREATE", "MATERIALIZED", "VIEW", MatchAny))
+ /* Complete CREATE MATERIALIZED VIEW <name> with AS */
+ else if (Matches("CREATE", "MATERIALIZED", "VIEW", MatchAny) ||
+ Matches("CREATE", "INCREMENTAL", "MATERIALIZED", "VIEW", MatchAny))
COMPLETE_WITH("AS");
/* Complete "CREATE MATERIALIZED VIEW <sth> AS with "SELECT" */
- else if (Matches("CREATE", "MATERIALIZED", "VIEW", MatchAny, "AS"))
+ else if (Matches("CREATE", "MATERIALIZED", "VIEW", MatchAny, "AS") ||
+ Matches("CREATE", "INCREMENTAL", "MATERIALIZED", "VIEW", MatchAny, "AS"))
COMPLETE_WITH("SELECT");
/* CREATE EVENT TRIGGER */
diff --git a/src/include/catalog/pg_class.dat b/src/include/catalog/pg_class.dat
index 9bcf28676d..7deda405af 100644
--- a/src/include/catalog/pg_class.dat
+++ b/src/include/catalog/pg_class.dat
@@ -27,7 +27,7 @@
relpersistence => 'p', relkind => 'r', relnatts => '31', relchecks => '0',
relhasrules => 'f', relhastriggers => 'f', relhassubclass => 'f',
relrowsecurity => 'f', relforcerowsecurity => 'f', relispopulated => 't',
- relreplident => 'n', relispartition => 'f', relfrozenxid => '3',
+ relreplident => 'n', relispartition => 'f', relisivm => 'f', relfrozenxid => '3',
relminmxid => '1', relacl => '_null_', reloptions => '_null_',
relpartbound => '_null_' },
{ oid => '1249',
@@ -37,7 +37,7 @@
relpersistence => 'p', relkind => 'r', relnatts => '25', relchecks => '0',
relhasrules => 'f', relhastriggers => 'f', relhassubclass => 'f',
relrowsecurity => 'f', relforcerowsecurity => 'f', relispopulated => 't',
- relreplident => 'n', relispartition => 'f', relfrozenxid => '3',
+ relreplident => 'n', relispartition => 'f', relisivm => 'f', relfrozenxid => '3',
relminmxid => '1', relacl => '_null_', reloptions => '_null_',
relpartbound => '_null_' },
{ oid => '1255',
@@ -47,17 +47,17 @@
relpersistence => 'p', relkind => 'r', relnatts => '29', relchecks => '0',
relhasrules => 'f', relhastriggers => 'f', relhassubclass => 'f',
relrowsecurity => 'f', relforcerowsecurity => 'f', relispopulated => 't',
- relreplident => 'n', relispartition => 'f', relfrozenxid => '3',
+ relreplident => 'n', relispartition => 'f', relisivm => 'f', relfrozenxid => '3',
relminmxid => '1', relacl => '_null_', reloptions => '_null_',
relpartbound => '_null_' },
{ oid => '1259',
relname => 'pg_class', reltype => 'pg_class', relam => 'heap',
relfilenode => '0', relpages => '0', reltuples => '0', relallvisible => '0',
reltoastrelid => '0', relhasindex => 'f', relisshared => 'f',
- relpersistence => 'p', relkind => 'r', relnatts => '33', relchecks => '0',
+ relpersistence => 'p', relkind => 'r', relnatts => '34', relchecks => '0',
relhasrules => 'f', relhastriggers => 'f', relhassubclass => 'f',
relrowsecurity => 'f', relforcerowsecurity => 'f', relispopulated => 't',
- relreplident => 'n', relispartition => 'f', relfrozenxid => '3',
+ relreplident => 'n', relispartition => 'f', relisivm => 'f', relfrozenxid => '3',
relminmxid => '1', relacl => '_null_', reloptions => '_null_',
relpartbound => '_null_' },
diff --git a/src/include/catalog/pg_class.h b/src/include/catalog/pg_class.h
index 090b6ba907..ff535f5504 100644
--- a/src/include/catalog/pg_class.h
+++ b/src/include/catalog/pg_class.h
@@ -116,6 +116,9 @@ CATALOG(pg_class,1259,RelationRelationId) BKI_BOOTSTRAP BKI_ROWTYPE_OID(83,Relat
/* is relation a partition? */
bool relispartition;
+ /* is relation a matview with ivm? */
+ bool relisivm;
+
/* heap for rewrite during DDL, link to original rel */
Oid relrewrite BKI_DEFAULT(0);
diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat
index 58ea5b982b..d0ad20f89c 100644
--- a/src/include/catalog/pg_proc.dat
+++ b/src/include/catalog/pg_proc.dat
@@ -10731,4 +10731,13 @@
proname => 'pg_partition_root', prorettype => 'regclass',
proargtypes => 'regclass', prosrc => 'pg_partition_root' },
+# IVM
+{ oid => '786', descr => 'ivm trigger (before)',
+ proname => 'IVM_immediate_before', provolatile => 'v', prorettype => 'trigger',
+ proargtypes => '', prosrc => 'IVM_immediate_before' },
+{ oid => '787', descr => 'ivm trigger (after)',
+ proname => 'IVM_immediate_maintenance', provolatile => 'v', prorettype => 'trigger',
+ proargtypes => '', prosrc => 'IVM_immediate_maintenance' },
+
+
]
diff --git a/src/include/commands/matview.h b/src/include/commands/matview.h
index edf04bf415..edf6ddaebe 100644
--- a/src/include/commands/matview.h
+++ b/src/include/commands/matview.h
@@ -23,6 +23,8 @@
extern void SetMatViewPopulatedState(Relation relation, bool newstate);
+extern void SetMatViewIVMState(Relation relation, bool newstate);
+
extern ObjectAddress ExecRefreshMatView(RefreshMatViewStmt *stmt, const char *queryString,
ParamListInfo params, char *completionTag);
@@ -30,4 +32,8 @@ extern DestReceiver *CreateTransientRelDestReceiver(Oid oid);
extern bool MatViewIncrementalMaintenanceIsEnabled(void);
+extern Datum IVM_immediate_before(PG_FUNCTION_ARGS);
+extern Datum IVM_immediate_maintenance(PG_FUNCTION_ARGS);
+extern void AtAbort_IVM(void);
+
#endif /* MATVIEW_H */
diff --git a/src/include/commands/trigger.h b/src/include/commands/trigger.h
index a46feeedb0..3e63733d3e 100644
--- a/src/include/commands/trigger.h
+++ b/src/include/commands/trigger.h
@@ -253,6 +253,8 @@ extern void AfterTriggerEndSubXact(bool isCommit);
extern void AfterTriggerSetState(ConstraintsSetStmt *stmt);
extern bool AfterTriggerPendingOnRel(Oid relid);
+extern void SetTransitionTablePreserved(Oid relid, CmdType cmdType);
+
/*
* in utils/adt/ri_triggers.c
diff --git a/src/include/nodes/parsenodes.h b/src/include/nodes/parsenodes.h
index d93a79a554..637d9bacb9 100644
--- a/src/include/nodes/parsenodes.h
+++ b/src/include/nodes/parsenodes.h
@@ -1002,6 +1002,7 @@ typedef struct RangeTblEntry
char relkind; /* relation kind (see pg_class.relkind) */
int rellockmode; /* lock level that query requires on the rel */
struct TableSampleClause *tablesample; /* sampling info, or NULL */
+ bool relisivm;
/*
* Fields valid for a subquery RTE (else NULL):
@@ -2059,6 +2060,7 @@ typedef struct CreateStmt
char *tablespacename; /* table space to use, or NULL */
char *accessMethod; /* table access method */
bool if_not_exists; /* just do nothing if it already exists? */
+ bool ivm; /* incremental view maintenance is used by materialized view */
} CreateStmt;
/* ----------
diff --git a/src/include/nodes/primnodes.h b/src/include/nodes/primnodes.h
index 860a84de7c..6df58e2ff9 100644
--- a/src/include/nodes/primnodes.h
+++ b/src/include/nodes/primnodes.h
@@ -117,6 +117,7 @@ typedef struct IntoClause
char *tableSpaceName; /* table space to use, or NULL */
Node *viewQuery; /* materialized view's SELECT query */
bool skipData; /* true for WITH NO DATA */
+ bool ivm; /* true for WITH IVM */
} IntoClause;
diff --git a/src/include/parser/kwlist.h b/src/include/parser/kwlist.h
index 00ace8425e..d682ee11cc 100644
--- a/src/include/parser/kwlist.h
+++ b/src/include/parser/kwlist.h
@@ -198,6 +198,7 @@ PG_KEYWORD("in", IN_P, RESERVED_KEYWORD)
PG_KEYWORD("include", INCLUDE, UNRESERVED_KEYWORD)
PG_KEYWORD("including", INCLUDING, UNRESERVED_KEYWORD)
PG_KEYWORD("increment", INCREMENT, UNRESERVED_KEYWORD)
+PG_KEYWORD("incremental", INCREMENTAL, UNRESERVED_KEYWORD)
PG_KEYWORD("index", INDEX, UNRESERVED_KEYWORD)
PG_KEYWORD("indexes", INDEXES, UNRESERVED_KEYWORD)
PG_KEYWORD("inherit", INHERIT, UNRESERVED_KEYWORD)
diff --git a/src/include/utils/lsyscache.h b/src/include/utils/lsyscache.h
index c8df5bff9f..8fd919349e 100644
--- a/src/include/utils/lsyscache.h
+++ b/src/include/utils/lsyscache.h
@@ -129,6 +129,7 @@ extern Oid get_rel_namespace(Oid relid);
extern Oid get_rel_type_id(Oid relid);
extern char get_rel_relkind(Oid relid);
extern bool get_rel_relispartition(Oid relid);
+extern bool get_rel_relisivm(Oid relid);
extern Oid get_rel_tablespace(Oid relid);
extern char get_rel_persistence(Oid relid);
extern Oid get_transform_fromsql(Oid typid, Oid langid, List *trftypes);
diff --git a/src/include/utils/rel.h b/src/include/utils/rel.h
index a5cf804f9f..3322eb0794 100644
--- a/src/include/utils/rel.h
+++ b/src/include/utils/rel.h
@@ -567,6 +567,8 @@ typedef struct ViewOptions
*/
#define RelationIsPopulated(relation) ((relation)->rd_rel->relispopulated)
+#define RelationIsIVM(relation) ((relation)->rd_rel->relisivm)
+
/*
* RelationIsAccessibleInLogicalDecoding
* True if we need to log enough information to have access via
diff --git a/src/test/regress/expected/incremental_matview.out b/src/test/regress/expected/incremental_matview.out
new file mode 100644
index 0000000000..153b4401e1
--- /dev/null
+++ b/src/test/regress/expected/incremental_matview.out
@@ -0,0 +1,450 @@
+-- create a table to use as a basis for views and materialized views in various combinations
+CREATE TABLE mv_base_a (i int, j int);
+INSERT INTO mv_base_a VALUES
+ (1,10),
+ (2,20),
+ (3,30),
+ (4,40),
+ (5,50);
+CREATE TABLE mv_base_b (i int, k int);
+INSERT INTO mv_base_b VALUES
+ (1,101),
+ (2,102),
+ (3,103),
+ (4,104);
+CREATE INCREMENTAL MATERIALIZED VIEW mv_ivm_1 AS SELECT i,j,k FROM mv_base_a a INNER JOIN mv_base_b b USING(i);
+SELECT * FROM mv_ivm_1 ORDER BY 1,2,3;
+ i | j | k
+---+----+-----
+ 1 | 10 | 101
+ 2 | 20 | 102
+ 3 | 30 | 103
+ 4 | 40 | 104
+(4 rows)
+
+-- immediaite maintenance
+BEGIN;
+INSERT INTO mv_base_b VALUES(5,105);
+SELECT * FROM mv_ivm_1 ORDER BY 1,2,3;
+ i | j | k
+---+----+-----
+ 1 | 10 | 101
+ 2 | 20 | 102
+ 3 | 30 | 103
+ 4 | 40 | 104
+ 5 | 50 | 105
+(5 rows)
+
+UPDATE mv_base_a SET j = 0 WHERE i = 1;
+SELECT * FROM mv_ivm_1 ORDER BY 1,2,3;
+ i | j | k
+---+----+-----
+ 1 | 0 | 101
+ 2 | 20 | 102
+ 3 | 30 | 103
+ 4 | 40 | 104
+ 5 | 50 | 105
+(5 rows)
+
+DELETE FROM mv_base_b WHERE (i,k) = (5,105);
+SELECT * FROM mv_ivm_1 ORDER BY 1,2,3;
+ i | j | k
+---+----+-----
+ 1 | 0 | 101
+ 2 | 20 | 102
+ 3 | 30 | 103
+ 4 | 40 | 104
+(4 rows)
+
+ROLLBACK;
+SELECT * FROM mv_ivm_1 ORDER BY 1,2,3;
+ i | j | k
+---+----+-----
+ 1 | 10 | 101
+ 2 | 20 | 102
+ 3 | 30 | 103
+ 4 | 40 | 104
+(4 rows)
+
+-- result of materliazied view have DISTINCT clause or the duplicate result.
+BEGIN;
+CREATE INCREMENTAL MATERIALIZED VIEW mv_ivm_duplicate AS SELECT j FROM mv_base_a;
+CREATE INCREMENTAL MATERIALIZED VIEW mv_ivm_distinct AS SELECT DISTINCT j FROM mv_base_a;
+INSERT INTO mv_base_a VALUES(6,20);
+SELECT * FROM mv_ivm_duplicate ORDER BY 1;
+ j
+----
+ 10
+ 20
+ 20
+ 30
+ 40
+ 50
+(6 rows)
+
+SELECT * FROM mv_ivm_distinct ORDER BY 1;
+ j
+----
+ 10
+ 20
+ 30
+ 40
+ 50
+(5 rows)
+
+DELETE FROM mv_base_a WHERE (i,j) = (2,20);
+SELECT * FROM mv_ivm_duplicate ORDER BY 1;
+ j
+----
+ 10
+ 20
+ 30
+ 40
+ 50
+(5 rows)
+
+SELECT * FROM mv_ivm_distinct ORDER BY 1;
+ j
+----
+ 10
+ 20
+ 30
+ 40
+ 50
+(5 rows)
+
+ROLLBACK;
+-- support SUM(), COUNT() and AVG() aggregation function
+BEGIN;
+CREATE INCREMENTAL MATERIALIZED VIEW mv_ivm_agg AS SELECT i, SUM(j), COUNT(i),AVG(j) FROM mv_base_a GROUP BY i;
+SELECT * FROM mv_ivm_agg ORDER BY 1,2,3,4;
+ i | sum | count | avg
+---+-----+-------+---------------------
+ 1 | 10 | 1 | 10.0000000000000000
+ 2 | 20 | 1 | 20.0000000000000000
+ 3 | 30 | 1 | 30.0000000000000000
+ 4 | 40 | 1 | 40.0000000000000000
+ 5 | 50 | 1 | 50.0000000000000000
+(5 rows)
+
+INSERT INTO mv_base_a VALUES(2,100);
+SELECT * FROM mv_ivm_agg ORDER BY 1,2,3,4;
+ i | sum | count | avg
+---+-----+-------+---------------------
+ 1 | 10 | 1 | 10.0000000000000000
+ 2 | 120 | 2 | 60.0000000000000000
+ 3 | 30 | 1 | 30.0000000000000000
+ 4 | 40 | 1 | 40.0000000000000000
+ 5 | 50 | 1 | 50.0000000000000000
+(5 rows)
+
+UPDATE mv_base_a SET j = 200 WHERE (i,j) = (2,100);
+SELECT * FROM mv_ivm_agg ORDER BY 1,2,3,4;
+ i | sum | count | avg
+---+-----+-------+----------------------
+ 1 | 10 | 1 | 10.0000000000000000
+ 2 | 220 | 2 | 110.0000000000000000
+ 3 | 30 | 1 | 30.0000000000000000
+ 4 | 40 | 1 | 40.0000000000000000
+ 5 | 50 | 1 | 50.0000000000000000
+(5 rows)
+
+DELETE FROM mv_base_a WHERE (i,j) = (2,200);
+SELECT * FROM mv_ivm_agg ORDER BY 1,2,3,4;
+ i | sum | count | avg
+---+-----+-------+---------------------
+ 1 | 10 | 1 | 10.0000000000000000
+ 2 | 20 | 1 | 20.0000000000000000
+ 3 | 30 | 1 | 30.0000000000000000
+ 4 | 40 | 1 | 40.0000000000000000
+ 5 | 50 | 1 | 50.0000000000000000
+(5 rows)
+
+ROLLBACK;
+-- support COUNT(*) aggregation function
+BEGIN;
+CREATE INCREMENTAL MATERIALIZED VIEW mv_ivm_agg AS SELECT i, SUM(j),COUNT(*) FROM mv_base_a GROUP BY i;
+SELECT * FROM mv_ivm_agg ORDER BY 1,2,3;
+ i | sum | count
+---+-----+-------
+ 1 | 10 | 1
+ 2 | 20 | 1
+ 3 | 30 | 1
+ 4 | 40 | 1
+ 5 | 50 | 1
+(5 rows)
+
+INSERT INTO mv_base_a VALUES(2,100);
+SELECT * FROM mv_ivm_agg ORDER BY 1,2,3;
+ i | sum | count
+---+-----+-------
+ 1 | 10 | 1
+ 2 | 120 | 2
+ 3 | 30 | 1
+ 4 | 40 | 1
+ 5 | 50 | 1
+(5 rows)
+
+ROLLBACK;
+-- support having only aggregation function without GROUP clause
+BEGIN;
+CREATE INCREMENTAL MATERIALIZED VIEW mv_ivm_group AS SELECT SUM(j)FROM mv_base_a;
+SELECT * FROM mv_ivm_group ORDER BY 1;
+ sum
+-----
+ 150
+(1 row)
+
+INSERT INTO mv_base_a VALUES(6,20);
+SELECT * FROM mv_ivm_group ORDER BY 1;
+ sum
+-----
+ 170
+(1 row)
+
+ROLLBACK;
+-- resolved issue: When use AVG() function and values is indivisible, result of AVG() is incorrect.
+BEGIN;
+CREATE INCREMENTAL MATERIALIZED VIEW mv_ivm_avg_bug AS SELECT i, SUM(j), COUNT(j), AVG(j) FROM mv_base_A GROUP BY i;
+SELECT * FROM mv_ivm_avg_bug ORDER BY 1,2,3;
+ i | sum | count | avg
+---+-----+-------+---------------------
+ 1 | 10 | 1 | 10.0000000000000000
+ 2 | 20 | 1 | 20.0000000000000000
+ 3 | 30 | 1 | 30.0000000000000000
+ 4 | 40 | 1 | 40.0000000000000000
+ 5 | 50 | 1 | 50.0000000000000000
+(5 rows)
+
+INSERT INTO mv_base_a VALUES
+ (1,0),
+ (1,0),
+ (2,30),
+ (2,30);
+SELECT * FROM mv_ivm_avg_bug ORDER BY 1,2,3;
+ i | sum | count | avg
+---+-----+-------+---------------------
+ 1 | 10 | 3 | 3.3333333333333333
+ 2 | 80 | 3 | 26.6666666666666667
+ 3 | 30 | 1 | 30.0000000000000000
+ 4 | 40 | 1 | 40.0000000000000000
+ 5 | 50 | 1 | 50.0000000000000000
+(5 rows)
+
+DELETE FROM mv_base_a WHERE (i,j) = (1,0);
+DELETE FROM mv_base_a WHERE (i,j) = (2,30);
+SELECT * FROM mv_ivm_avg_bug ORDER BY 1,2,3;
+ i | sum | count | avg
+---+-----+-------+---------------------
+ 1 | 10 | 1 | 10.0000000000000000
+ 2 | 20 | 1 | 20.0000000000000000
+ 3 | 30 | 1 | 30.0000000000000000
+ 4 | 40 | 1 | 40.0000000000000000
+ 5 | 50 | 1 | 50.0000000000000000
+(5 rows)
+
+ROLLBACK;
+-- support MIN(), MAX() aggregation functions
+BEGIN;
+CREATE INCREMENTAL MATERIALIZED VIEW mv_ivm_min_max AS SELECT i, MIN(j), MAX(j) FROM mv_base_a GROUP BY i;
+SELECT * FROM mv_ivm_min_max ORDER BY 1,2,3;
+ i | min | max
+---+-----+-----
+ 1 | 10 | 10
+ 2 | 20 | 20
+ 3 | 30 | 30
+ 4 | 40 | 40
+ 5 | 50 | 50
+(5 rows)
+
+INSERT INTO mv_base_a VALUES
+ (1,11), (1,12),
+ (2,21), (2,22),
+ (3,31), (3,32),
+ (4,41), (4,42),
+ (5,51), (5,52);
+SELECT * FROM mv_ivm_min_max ORDER BY 1,2,3;
+ i | min | max
+---+-----+-----
+ 1 | 10 | 12
+ 2 | 20 | 22
+ 3 | 30 | 32
+ 4 | 40 | 42
+ 5 | 50 | 52
+(5 rows)
+
+DELETE FROM mv_base_a WHERE (i,j) IN ((1,10), (2,21), (3,32));
+SELECT * FROM mv_ivm_min_max ORDER BY 1,2,3;
+ i | min | max
+---+-----+-----
+ 1 | 11 | 12
+ 2 | 20 | 22
+ 3 | 30 | 31
+ 4 | 40 | 42
+ 5 | 50 | 52
+(5 rows)
+
+ROLLBACK;
+-- support MIN(), MAX() aggregation functions without GROUP clause
+BEGIN;
+CREATE INCREMENTAL MATERIALIZED VIEW mv_ivm_min_max AS SELECT MIN(j), MAX(j) FROM mv_base_a;
+SELECT * FROM mv_ivm_min_max;
+ min | max
+-----+-----
+ 10 | 50
+(1 row)
+
+INSERT INTO mv_base_a VALUES
+ (0,0), (6,60), (7,70);
+SELECT * FROM mv_ivm_min_max;
+ min | max
+-----+-----
+ 0 | 70
+(1 row)
+
+DELETE FROM mv_base_a WHERE (i,j) IN ((0,0), (7,70));
+SELECT * FROM mv_ivm_min_max;
+ min | max
+-----+-----
+ 10 | 60
+(1 row)
+
+ROLLBACK;
+-- support self join view and multiple change on the same table
+BEGIN;
+CREATE TABLE t (i int, v int);
+INSERT INTO t VALUES (1, 10), (2, 20), (3, 30);
+CREATE INCREMENTAL MATERIALIZED VIEW mv_self(v1, v2) AS
+ SELECT t1.v, t2.v FROM t t1 JOIN t t2 ON t1.i = t2.i;
+SELECT * FROM mv_self ORDER BY v1;
+ v1 | v2
+----+----
+ 10 | 10
+ 20 | 20
+ 30 | 30
+(3 rows)
+
+INSERT INTO t VALUES (4,40);
+DELETE FROM t WHERE i = 1;
+UPDATE t SET v = v*10 WHERE i=2;
+SELECT * FROM mv_self ORDER BY v1;
+ v1 | v2
+-----+-----
+ 30 | 30
+ 40 | 40
+ 200 | 200
+(3 rows)
+
+WITH
+ ins_t1 AS (INSERT INTO t VALUES (5,50) RETURNING 1),
+ ins_t2 AS (INSERT INTO t VALUES (6,60) RETURNING 1),
+ upd_t AS (UPDATE t SET v = v + 100 RETURNING 1),
+ dlt_t AS (DELETE FROM t WHERE i IN (4,5) RETURNING 1)
+SELECT NULL;
+ ?column?
+----------
+
+(1 row)
+
+SELECT * FROM mv_self ORDER BY v1;
+ v1 | v2
+-----+-----
+ 50 | 50
+ 60 | 60
+ 130 | 130
+ 300 | 300
+(4 rows)
+
+ROLLBACK;
+-- support simultaneous table changes
+BEGIN;
+CREATE TABLE r (i int, v int);
+CREATE TABLE s (i int, v int);
+INSERT INTO r VALUES (1, 10), (2, 20), (3, 30);
+INSERT INTO s VALUES (1, 100), (2, 200), (3, 300);
+CREATE INCREMENTAL MATERIALIZED VIEW mv(v1, v2) AS
+ SELECT r.v, s.v FROM r JOIN s USING(i);
+SELECT * FROM mv ORDER BY v1;
+ v1 | v2
+----+-----
+ 10 | 100
+ 20 | 200
+ 30 | 300
+(3 rows)
+
+WITH
+ ins_r AS (INSERT INTO r VALUES (1,11) RETURNING 1),
+ ins_r2 AS (INSERT INTO r VALUES (3,33) RETURNING 1),
+ ins_s AS (INSERT INTO s VALUES (2,222) RETURNING 1),
+ upd_r AS (UPDATE r SET v = v + 1000 WHERE i = 2 RETURNING 1),
+ dlt_s AS (DELETE FROM s WHERE i = 3 RETURNING 1)
+SELECT NULL;
+ ?column?
+----------
+
+(1 row)
+
+SELECT * FROM mv ORDER BY v1;
+ v1 | v2
+------+-----
+ 10 | 100
+ 11 | 100
+ 1020 | 200
+ 1020 | 222
+(4 rows)
+
+ROLLBACK;
+-- support foreign reference constrains
+BEGIN;
+CREATE TABLE ri1 (i int PRIMARY KEY);
+CREATE TABLE ri2 (i int PRIMARY KEY REFERENCES ri1(i) ON UPDATE CASCADE ON DELETE CASCADE, v int);
+INSERT INTO ri1 VALUES (1),(2),(3);
+INSERT INTO ri2 VALUES (1),(2),(3);
+CREATE INCREMENTAL MATERIALIZED VIEW mv_ri(i1, i2) AS
+ SELECT ri1.i, ri2.i FROM ri1 JOIN ri2 USING(i);
+SELECT * FROM mv_ri ORDER BY i1;
+ i1 | i2
+----+----
+ 1 | 1
+ 2 | 2
+ 3 | 3
+(3 rows)
+
+UPDATE ri1 SET i=10 where i=1;
+DELETE FROM ri1 WHERE i=2;
+SELECT * FROM mv_ri ORDER BY i2;
+ i1 | i2
+----+----
+ 3 | 3
+ 10 | 10
+(2 rows)
+
+ROLLBACK;
+-- contain system column
+CREATE INCREMENTAL MATERIALIZED VIEW mv_ivm01 AS SELECT i,j,xmin FROM mv_base_a;
+ERROR: system column is not supported with IVM
+CREATE INCREMENTAL MATERIALIZED VIEW mv_ivm02 AS SELECT i,j FROM mv_base_a WHERE xmin = '610';
+ERROR: system column is not supported with IVM
+-- contain subquery
+CREATE INCREMENTAL MATERIALIZED VIEW mv_ivm03 AS SELECT i,j FROM mv_base_a WHERE i IN (SELECT i FROM mv_base_b WHERE k < 103 );
+ERROR: subquery is not supported with IVM
+CREATE INCREMENTAL MATERIALIZED VIEW mv_ivm04 AS SELECT a.i,a.j FROM mv_base_a a,( SELECT * FROM mv_base_b) b WHERE a.i = b.i;
+ERROR: subquery is not supported with IVM
+CREATE INCREMENTAL MATERIALIZED VIEW mv_ivm05 AS SELECT i,j, (SELECT k FROM mv_base_b b WHERE a.i = b.i) FROM mv_base_a a;
+ERROR: subquery is not supported with IVM
+-- contain CTE
+CREATE INCREMENTAL MATERIALIZED VIEW mv_ivm06 AS WITH b AS (SELECT i,k FROM mv_base_b WHERE k < 103) SELECT a.i,a.j FROM mv_base_a a,b WHERE a.i = b.i;
+ERROR: CTE is not supported with IVM
+-- contain view or materialized view
+CREATE VIEW b_view AS SELECT i,k FROM mv_base_b;
+CREATE MATERIALIZED VIEW b_mview AS SELECT i,k FROM mv_base_b;
+CREATE INCREMENTAL MATERIALIZED VIEW mv_ivm07 AS SELECT a.i,a.j FROM mv_base_a a,b_view b WHERE a.i = b.i;
+ERROR: VIEW or MATERIALIZED VIEW is not supported with IVM
+CREATE INCREMENTAL MATERIALIZED VIEW mv_ivm08 AS SELECT a.i,a.j FROM mv_base_a a,b_mview b WHERE a.i = b.i;
+ERROR: VIEW or MATERIALIZED VIEW is not supported with IVM
+DROP TABLE mv_base_b CASCADE;
+NOTICE: drop cascades to 3 other objects
+DETAIL: drop cascades to materialized view mv_ivm_1
+drop cascades to view b_view
+drop cascades to materialized view b_mview
+DROP TABLE mv_base_a CASCADE;
diff --git a/src/test/regress/parallel_schedule b/src/test/regress/parallel_schedule
index fc0f14122b..4b5d448098 100644
--- a/src/test/regress/parallel_schedule
+++ b/src/test/regress/parallel_schedule
@@ -78,7 +78,7 @@ test: brin gin gist spgist privileges init_privs security_label collate matview
# ----------
# Another group of parallel tests
# ----------
-test: create_table_like alter_generic alter_operator misc async dbsize misc_functions sysviews tsrf tidscan collate.icu.utf8
+test: create_table_like alter_generic alter_operator misc async dbsize misc_functions sysviews tsrf tidscan collate.icu.utf8 incremental_matview
# rules cannot run concurrently with any test that creates
# a view or rule in the public schema
diff --git a/src/test/regress/serial_schedule b/src/test/regress/serial_schedule
index 68ac56acdb..0f6f8f503f 100644
--- a/src/test/regress/serial_schedule
+++ b/src/test/regress/serial_schedule
@@ -112,6 +112,7 @@ test: init_privs
test: security_label
test: collate
test: matview
+test: incremental_matview
test: lock
test: replica_identity
test: rowsecurity
diff --git a/src/test/regress/sql/incremental_matview.sql b/src/test/regress/sql/incremental_matview.sql
new file mode 100644
index 0000000000..d815bd8662
--- /dev/null
+++ b/src/test/regress/sql/incremental_matview.sql
@@ -0,0 +1,180 @@
+-- create a table to use as a basis for views and materialized views in various combinations
+CREATE TABLE mv_base_a (i int, j int);
+INSERT INTO mv_base_a VALUES
+ (1,10),
+ (2,20),
+ (3,30),
+ (4,40),
+ (5,50);
+CREATE TABLE mv_base_b (i int, k int);
+INSERT INTO mv_base_b VALUES
+ (1,101),
+ (2,102),
+ (3,103),
+ (4,104);
+
+CREATE INCREMENTAL MATERIALIZED VIEW mv_ivm_1 AS SELECT i,j,k FROM mv_base_a a INNER JOIN mv_base_b b USING(i);
+
+SELECT * FROM mv_ivm_1 ORDER BY 1,2,3;
+-- immediaite maintenance
+BEGIN;
+INSERT INTO mv_base_b VALUES(5,105);
+SELECT * FROM mv_ivm_1 ORDER BY 1,2,3;
+UPDATE mv_base_a SET j = 0 WHERE i = 1;
+SELECT * FROM mv_ivm_1 ORDER BY 1,2,3;
+DELETE FROM mv_base_b WHERE (i,k) = (5,105);
+SELECT * FROM mv_ivm_1 ORDER BY 1,2,3;
+ROLLBACK;
+SELECT * FROM mv_ivm_1 ORDER BY 1,2,3;
+
+-- result of materliazied view have DISTINCT clause or the duplicate result.
+BEGIN;
+CREATE INCREMENTAL MATERIALIZED VIEW mv_ivm_duplicate AS SELECT j FROM mv_base_a;
+CREATE INCREMENTAL MATERIALIZED VIEW mv_ivm_distinct AS SELECT DISTINCT j FROM mv_base_a;
+INSERT INTO mv_base_a VALUES(6,20);
+SELECT * FROM mv_ivm_duplicate ORDER BY 1;
+SELECT * FROM mv_ivm_distinct ORDER BY 1;
+DELETE FROM mv_base_a WHERE (i,j) = (2,20);
+SELECT * FROM mv_ivm_duplicate ORDER BY 1;
+SELECT * FROM mv_ivm_distinct ORDER BY 1;
+ROLLBACK;
+
+-- support SUM(), COUNT() and AVG() aggregation function
+BEGIN;
+CREATE INCREMENTAL MATERIALIZED VIEW mv_ivm_agg AS SELECT i, SUM(j), COUNT(i),AVG(j) FROM mv_base_a GROUP BY i;
+SELECT * FROM mv_ivm_agg ORDER BY 1,2,3,4;
+INSERT INTO mv_base_a VALUES(2,100);
+SELECT * FROM mv_ivm_agg ORDER BY 1,2,3,4;
+UPDATE mv_base_a SET j = 200 WHERE (i,j) = (2,100);
+SELECT * FROM mv_ivm_agg ORDER BY 1,2,3,4;
+DELETE FROM mv_base_a WHERE (i,j) = (2,200);
+SELECT * FROM mv_ivm_agg ORDER BY 1,2,3,4;
+ROLLBACK;
+
+-- support COUNT(*) aggregation function
+BEGIN;
+CREATE INCREMENTAL MATERIALIZED VIEW mv_ivm_agg AS SELECT i, SUM(j),COUNT(*) FROM mv_base_a GROUP BY i;
+SELECT * FROM mv_ivm_agg ORDER BY 1,2,3;
+INSERT INTO mv_base_a VALUES(2,100);
+SELECT * FROM mv_ivm_agg ORDER BY 1,2,3;
+ROLLBACK;
+
+-- support having only aggregation function without GROUP clause
+BEGIN;
+CREATE INCREMENTAL MATERIALIZED VIEW mv_ivm_group AS SELECT SUM(j)FROM mv_base_a;
+SELECT * FROM mv_ivm_group ORDER BY 1;
+INSERT INTO mv_base_a VALUES(6,20);
+SELECT * FROM mv_ivm_group ORDER BY 1;
+ROLLBACK;
+
+-- resolved issue: When use AVG() function and values is indivisible, result of AVG() is incorrect.
+BEGIN;
+CREATE INCREMENTAL MATERIALIZED VIEW mv_ivm_avg_bug AS SELECT i, SUM(j), COUNT(j), AVG(j) FROM mv_base_A GROUP BY i;
+SELECT * FROM mv_ivm_avg_bug ORDER BY 1,2,3;
+INSERT INTO mv_base_a VALUES
+ (1,0),
+ (1,0),
+ (2,30),
+ (2,30);
+SELECT * FROM mv_ivm_avg_bug ORDER BY 1,2,3;
+DELETE FROM mv_base_a WHERE (i,j) = (1,0);
+DELETE FROM mv_base_a WHERE (i,j) = (2,30);
+SELECT * FROM mv_ivm_avg_bug ORDER BY 1,2,3;
+ROLLBACK;
+
+-- support MIN(), MAX() aggregation functions
+BEGIN;
+CREATE INCREMENTAL MATERIALIZED VIEW mv_ivm_min_max AS SELECT i, MIN(j), MAX(j) FROM mv_base_a GROUP BY i;
+SELECT * FROM mv_ivm_min_max ORDER BY 1,2,3;
+INSERT INTO mv_base_a VALUES
+ (1,11), (1,12),
+ (2,21), (2,22),
+ (3,31), (3,32),
+ (4,41), (4,42),
+ (5,51), (5,52);
+SELECT * FROM mv_ivm_min_max ORDER BY 1,2,3;
+DELETE FROM mv_base_a WHERE (i,j) IN ((1,10), (2,21), (3,32));
+SELECT * FROM mv_ivm_min_max ORDER BY 1,2,3;
+ROLLBACK;
+
+-- support MIN(), MAX() aggregation functions without GROUP clause
+BEGIN;
+CREATE INCREMENTAL MATERIALIZED VIEW mv_ivm_min_max AS SELECT MIN(j), MAX(j) FROM mv_base_a;
+SELECT * FROM mv_ivm_min_max;
+INSERT INTO mv_base_a VALUES
+ (0,0), (6,60), (7,70);
+SELECT * FROM mv_ivm_min_max;
+DELETE FROM mv_base_a WHERE (i,j) IN ((0,0), (7,70));
+SELECT * FROM mv_ivm_min_max;
+ROLLBACK;
+
+-- support self join view and multiple change on the same table
+BEGIN;
+CREATE TABLE t (i int, v int);
+INSERT INTO t VALUES (1, 10), (2, 20), (3, 30);
+CREATE INCREMENTAL MATERIALIZED VIEW mv_self(v1, v2) AS
+ SELECT t1.v, t2.v FROM t t1 JOIN t t2 ON t1.i = t2.i;
+SELECT * FROM mv_self ORDER BY v1;
+INSERT INTO t VALUES (4,40);
+DELETE FROM t WHERE i = 1;
+UPDATE t SET v = v*10 WHERE i=2;
+SELECT * FROM mv_self ORDER BY v1;
+WITH
+ ins_t1 AS (INSERT INTO t VALUES (5,50) RETURNING 1),
+ ins_t2 AS (INSERT INTO t VALUES (6,60) RETURNING 1),
+ upd_t AS (UPDATE t SET v = v + 100 RETURNING 1),
+ dlt_t AS (DELETE FROM t WHERE i IN (4,5) RETURNING 1)
+SELECT NULL;
+SELECT * FROM mv_self ORDER BY v1;
+ROLLBACK;
+
+-- support simultaneous table changes
+BEGIN;
+CREATE TABLE r (i int, v int);
+CREATE TABLE s (i int, v int);
+INSERT INTO r VALUES (1, 10), (2, 20), (3, 30);
+INSERT INTO s VALUES (1, 100), (2, 200), (3, 300);
+CREATE INCREMENTAL MATERIALIZED VIEW mv(v1, v2) AS
+ SELECT r.v, s.v FROM r JOIN s USING(i);
+SELECT * FROM mv ORDER BY v1;
+WITH
+ ins_r AS (INSERT INTO r VALUES (1,11) RETURNING 1),
+ ins_r2 AS (INSERT INTO r VALUES (3,33) RETURNING 1),
+ ins_s AS (INSERT INTO s VALUES (2,222) RETURNING 1),
+ upd_r AS (UPDATE r SET v = v + 1000 WHERE i = 2 RETURNING 1),
+ dlt_s AS (DELETE FROM s WHERE i = 3 RETURNING 1)
+SELECT NULL;
+SELECT * FROM mv ORDER BY v1;
+ROLLBACK;
+
+-- support foreign reference constrains
+BEGIN;
+CREATE TABLE ri1 (i int PRIMARY KEY);
+CREATE TABLE ri2 (i int PRIMARY KEY REFERENCES ri1(i) ON UPDATE CASCADE ON DELETE CASCADE, v int);
+INSERT INTO ri1 VALUES (1),(2),(3);
+INSERT INTO ri2 VALUES (1),(2),(3);
+CREATE INCREMENTAL MATERIALIZED VIEW mv_ri(i1, i2) AS
+ SELECT ri1.i, ri2.i FROM ri1 JOIN ri2 USING(i);
+SELECT * FROM mv_ri ORDER BY i1;
+UPDATE ri1 SET i=10 where i=1;
+DELETE FROM ri1 WHERE i=2;
+SELECT * FROM mv_ri ORDER BY i2;
+ROLLBACK;
+
+-- contain system column
+CREATE INCREMENTAL MATERIALIZED VIEW mv_ivm01 AS SELECT i,j,xmin FROM mv_base_a;
+CREATE INCREMENTAL MATERIALIZED VIEW mv_ivm02 AS SELECT i,j FROM mv_base_a WHERE xmin = '610';
+-- contain subquery
+CREATE INCREMENTAL MATERIALIZED VIEW mv_ivm03 AS SELECT i,j FROM mv_base_a WHERE i IN (SELECT i FROM mv_base_b WHERE k < 103 );
+CREATE INCREMENTAL MATERIALIZED VIEW mv_ivm04 AS SELECT a.i,a.j FROM mv_base_a a,( SELECT * FROM mv_base_b) b WHERE a.i = b.i;
+CREATE INCREMENTAL MATERIALIZED VIEW mv_ivm05 AS SELECT i,j, (SELECT k FROM mv_base_b b WHERE a.i = b.i) FROM mv_base_a a;
+-- contain CTE
+CREATE INCREMENTAL MATERIALIZED VIEW mv_ivm06 AS WITH b AS (SELECT i,k FROM mv_base_b WHERE k < 103) SELECT a.i,a.j FROM mv_base_a a,b WHERE a.i = b.i;
+-- contain view or materialized view
+CREATE VIEW b_view AS SELECT i,k FROM mv_base_b;
+CREATE MATERIALIZED VIEW b_mview AS SELECT i,k FROM mv_base_b;
+CREATE INCREMENTAL MATERIALIZED VIEW mv_ivm07 AS SELECT a.i,a.j FROM mv_base_a a,b_view b WHERE a.i = b.i;
+CREATE INCREMENTAL MATERIALIZED VIEW mv_ivm08 AS SELECT a.i,a.j FROM mv_base_a a,b_mview b WHERE a.i = b.i;
+
+DROP TABLE mv_base_b CASCADE;
+DROP TABLE mv_base_a CASCADE;
