Hi,
Attached is the latest patch for supporting min and max aggregate functions.
When new tuples are inserted into base tables, if new values are small
(for min) or large (for max), matview just have to be updated with these
new values. Otherwise, old values just remains.
However, in the case of deletion, this is more complicated. If deleted
values exists in matview as current min or max, we have to recomputate
new min or max values from base tables for affected groups, and matview
should be updated with these recomputated values.
Also, regression tests for min/max are also added.
In addition, incremental update algorithm of avg aggregate values is a bit
improved. If an avg result in materialized views is updated incrementally
y using the old avg value, numerical errors in avg values are accumulated
and the values get wrong eventually. To prevent this, both of sum and count
values are contained in views as hidden columns and use them to calculate
new avg value instead of using old avg values.
Regards,
On Fri, 26 Jul 2019 11:35:53 +0900
Yugo Nagata <nag...@sraoss.co.jp> wrote:
> Hi,
>
> I've updated the wiki page of Incremental View Maintenance.
>
> https://wiki.postgresql.org/wiki/Incremental_View_Maintenance
>
> On Thu, 11 Jul 2019 13:28:04 +0900
> Yugo Nagata <nag...@sraoss.co.jp> wrote:
>
> > Hi Thomas,
> >
> > Thank you for your review and discussion on this patch!
> >
> > > > 2019年7月8日(月) 15:32 Thomas Munro <thomas.mu...@gmail.com>:
> > > >
> > > > > On Fri, Jun 28, 2019 at 10:56 PM Yugo Nagata <nag...@sraoss.co.jp>
> > > > > wrote:
> > > > > > Attached is a WIP patch of IVM which supports some aggregate
> > > > > > functions.
> > > > >
> > > > > Hi Nagata-san and Hoshiai-san,
> > > > >
> > > > > Thank you for working on this. I enjoyed your talk at PGCon. I've
> > > > > added Kevin Grittner just in case he missed this thread; he has talked
> > > > > often about implementing the counting algorithm, and he wrote the
> > > > > "trigger transition tables" feature to support exactly this. While
> > > > > integrating trigger transition tables with the new partition features,
> > > > > we had to make a number of decisions about how that should work, and
> > > > > we tried to come up with answers that would work for IMV, and I hope
> > > > > we made the right choices!
> >
> > Transition tables is a great feature. I am now using this in my
> > implementation
> > of IVM, but the first time I used this feature was when I implemented a PoC
> > for extending view updatability of PostgreSQL[1]. At that time, I didn't
> > know
> > that this feature is made originally aiming to support IVM.
> >
> > [1] https://www.pgcon.org/2017/schedule/events/1074.en.html
> >
> > I think transition tables is a good choice to implement a statement level
> > immediate view maintenance where materialized views are refreshed in a
> > statement
> > level after trigger. However, when implementing a transaction level
> > immediate
> > view maintenance where views are refreshed per transaction, or deferred view
> > maintenance, we can't update views in a after trigger, and we will need an
> > infrastructure to manage change logs of base tables. Transition tables can
> > be
> > used to collect these logs, but using logical decoding of WAL is another
> > candidate.
> > In any way, if these logs can be collected in a tuplestore, we might able to
> > convert this to "ephemeral named relation (ENR)" and use this to calculate
> > diff
> > sets for views.
> >
> > > > >
> > > > > I am quite interested to learn how IVM interacts with SERIALIZABLE.
> > > > >
> > > >
> > > > Nagata-san has been studying this. Nagata-san, any comment?
> >
> > In SERIALIZABLE or REPEATABLE READ level, table changes occurred in other
> > ransactions are not visible, so views can not be maintained correctly in
> > AFTER
> > triggers. If a view is defined on two tables and each table is modified in
> > different concurrent transactions respectively, the result of view
> > maintenance
> > done in trigger functions can be incorrect due to the race condition. This
> > is the
> > reason why such transactions are aborted immediately in that case in my
> > current
> > implementation.
> >
> > One idea to resolve this is performing view maintenance in two phases.
> > Firstly,
> > views are updated using only table changes visible in this transaction.
> > Then,
> > just after this transaction is committed, views have to be updated
> > additionally
> > using changes happened in other transactions to keep consistency. This is a
> > just
> > idea, but to implement this idea, I think, we will need keep to keep and
> > maintain change logs.
> >
> > > > > A couple of superficial review comments:
> >
> >
> >
> > > > > + const char *aggname = get_func_name(aggref->aggfnoid);
> > > > > ...
> > > > > + else if (!strcmp(aggname, "sum"))
> > > > >
> > > > > I guess you need a more robust way to detect the supported aggregates
> > > > > than their name, or I guess some way for aggregates themselves to
> > > > > specify that they support this and somehow supply the extra logic.
> > > > > Perhaps I just waid what Greg Stark already said, except not as well.
> >
> > Yes. Using name is not robust because users can make same name aggregates
> > like
> > sum(text) (although I am not sure this makes sense). We can use oids
> > instead
> > of names, but it would be nice to extend pg_aggregate and add new
> > attributes
> > for informing that this supports IVM and for providing functions for IVM
> > logic.
> >
> > > > > As for the question of how
> > > > > to reserve a namespace for system columns that won't clash with user
> > > > > columns, according to our manual the SQL standard doesn't allow $ in
> > > > > identifier names, and according to my copy SQL92 "intermediate SQL"
> > > > > doesn't allow identifiers that end in an underscore. I don't know
> > > > > what the best answer is but we should probably decide on a something
> > > > > based the standard.
> >
> > Ok, so we should use "__ivm_count__" since this ends in "_" at least.
> >
> > Another idea is that users specify the name of this special column when
> > defining materialized views with IVM support. This way can avoid the
> > conflict
> > because users will specify a name which does not appear in the target list.
> >
> > As for aggregates supports, it may be also possible to make it a
> > restriction
> > that count(expr) must be in the target list explicitly when sum(expr) or
> > avg(expr) is included, instead of making hidden column like
> > __ivm_count_sum__,
> > like Oracle does.
> >
> > > > >
> > > > > As for how to make internal columns invisible to SELECT *, previously
> > > > > there have been discussions about doing that using a new flag in
> > > > > pg_attribute:
> > > > >
> > > > >
> > > > > https://www.postgresql.org/message-id/flat/CAEepm%3D3ZHh%3Dp0nEEnVbs1Dig_UShPzHUcMNAqvDQUgYgcDo-pA%40mail.gmail.com
> >
> > I agree implementing this feature in PostgreSQL since there are at least a
> > few
> > use cases, IVM and temporal database.
> >
> > > > >
> > > > > + "WITH t AS ("
> > > > > + " SELECT diff.__ivm_count__,
> > > > > (diff.__ivm_count__ = mv.__ivm_count__) AS for_dlt, mv.ctid"
> > > > > + ", %s"
> > > > > + " 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"
> > > > > + ") DELETE FROM %s AS mv USING t WHERE
> > > > > mv.ctid = t.ctid AND for_dlt;",
> > > > >
> > > > > I fully understand that this is POC code, but I am curious about one
> > > > > thing. These queries that are executed by apply_delta() would need to
> > > > > be converted to C, or at least used reusable plans, right? Hmm,
> > > > > creating and dropping temporary tables every time is a clue that the
> > > > > ultimate form of this should be tuplestores and C code, I think,
> > > > > right?
> >
> > I used SPI just because REFRESH CONCURRENTLY uses this, but, as you said,
> > it is inefficient to create/drop temp tables and perform parse/plan every
> > times.
> > It seems to be enough to perform this once when creating materialized views
> > or
> > at the first maintenance time.
> >
> >
> > Best regards,
> > Yugo Nagata
> >
> >
> > --
> > Yugo Nagata <nag...@sraoss.co.jp>
> >
> >
>
>
> --
> Yugo Nagata <nag...@sraoss.co.jp>
>
>
--
Yugo Nagata <nag...@sraoss.co.jp>
diff --git a/doc/src/sgml/catalogs.sgml b/doc/src/sgml/catalogs.sgml
index 68ad5071ca..cd1fff8409 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/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 99ae159f98..c3cbe32ac7 100644
--- a/src/backend/catalog/index.c
+++ b/src/backend/catalog/index.c
@@ -910,6 +910,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..1ff5d03d59 100644
--- a/src/backend/commands/createas.c
+++ b/src/backend/commands/createas.c
@@ -51,6 +51,17 @@
#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 "nodes/print.h"
+#include "nodes/primnodes.h"
+#include "optimizer/optimizer.h"
+#include "commands/defrem.h"
+
typedef struct
{
@@ -74,6 +85,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);
+static void CreateIvmTriggersOnBaseTables(Query *qry, Node *jtnode, Oid matviewOid, char* matviewname);
+static void check_ivm_restriction_walker(Node *node);
/*
* create_ctas_internal
@@ -109,6 +123,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 +255,7 @@ ExecCreateTableAs(CreateTableAsStmt *stmt, const char *queryString,
List *rewritten;
PlannedStmt *plan;
QueryDesc *queryDesc;
+ Query *copied_query;
if (stmt->if_not_exists)
{
@@ -256,6 +273,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 +339,130 @@ 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)
+ {
+ 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, list_make1(dmy_arg), 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 +521,65 @@ ExecCreateTableAs(CreateTableAsStmt *stmt, const char *queryString,
/* Restore userid and security context */
SetUserIdAndSecContext(save_userid, save_sec_context);
+
+
+ if (into->ivm)
+ {
+ char *matviewname;
+ Oid matviewOid = address.objectId;
+ Relation matviewRel = table_open(matviewOid, NoLock);
+ matviewname = quote_qualified_identifier(get_namespace_name(RelationGetNamespace(matviewRel)),
+ RelationGetRelationName(matviewRel));
+ copied_query = copyObject(query);
+ AcquireRewriteLocks(copied_query, true, false);
+
+ CreateIvmTriggersOnBaseTables(copied_query, (Node *)copied_query->jointree, matviewOid, matviewname);
+
+ table_close(matviewRel, NoLock);
+ }
}
return address;
}
+static void CreateIvmTriggersOnBaseTables(Query *qry, Node *jtnode, Oid matviewOid, char* matviewname)
+{
+
+ 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)
+ {
+ CreateIvmTrigger(rte->relid, matviewOid, matviewname, TRIGGER_TYPE_INSERT);
+ CreateIvmTrigger(rte->relid, matviewOid, matviewname, TRIGGER_TYPE_DELETE);
+ CreateIvmTrigger(rte->relid, matviewOid, matviewname, TRIGGER_TYPE_UPDATE);
+ }
+ 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);
+ }
+ else if (IsA(jtnode, JoinExpr))
+ {
+ JoinExpr *j = (JoinExpr *) jtnode;
+
+ CreateIvmTriggersOnBaseTables(qry, j->larg, matviewOid, matviewname);
+ CreateIvmTriggersOnBaseTables(qry, j->rarg, matviewOid, matviewname);
+ }
+ else
+ elog(ERROR, "unrecognized node type: %d", (int) nodeTag(jtnode));
+}
+
/*
* GetIntoRelEFlags --- compute executor flags needed for CREATE TABLE AS
*
@@ -547,6 +744,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 +821,229 @@ intorel_destroy(DestReceiver *self)
{
pfree(self);
}
+
+static void
+CreateIvmTrigger(Oid relOid, Oid viewOid, char *matviewname, int16 type)
+{
+ CreateTrigStmt *ivm_trigger;
+ List *transitionRels = NIL;
+ ObjectAddress address, refaddr;
+
+ refaddr.classId = RelationRelationId;
+ refaddr.objectId = viewOid;
+ refaddr.objectSubId = 0;
+
+
+ ivm_trigger = makeNode(CreateTrigStmt);
+ ivm_trigger->relation = NULL;
+ ivm_trigger->row = false;
+ ivm_trigger->timing = TRIGGER_TYPE_AFTER;
+
+ ivm_trigger->events = type;
+
+ switch (type)
+ {
+ case TRIGGER_TYPE_INSERT:
+ ivm_trigger->trigname = "IVM_trigger_ins";
+ break;
+ case TRIGGER_TYPE_DELETE:
+ ivm_trigger->trigname = "IVM_trigger_del";
+ break;
+ case TRIGGER_TYPE_UPDATE:
+ ivm_trigger->trigname = "IVM_trigger_upd";
+ break;
+ }
+
+ 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 = 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..be235d28ca 100644
--- a/src/backend/commands/matview.c
+++ b/src/backend/commands/matview.c
@@ -46,6 +46,15 @@
#include "utils/snapmgr.h"
#include "utils/syscache.h"
+#include "utils/regproc.h"
+#include "nodes/makefuncs.h"
+#include "parser/parse_clause.h"
+#include "parser/parse_func.h"
+#include "nodes/print.h"
+#include "catalog/pg_type_d.h"
+#include "optimizer/optimizer.h"
+#include "commands/defrem.h"
+
typedef struct
{
@@ -60,12 +69,16 @@ typedef struct
static int matview_maintenance_depth = 0;
+static SPIPlanPtr plan_for_recalc_min_max = NULL;
+static SPIPlanPtr plan_for_set_min_max = NULL;
+
static void transientrel_startup(DestReceiver *self, int operation, TupleDesc typeinfo);
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 +87,16 @@ static bool is_usable_unique_index(Relation indexRel);
static void OpenMatViewIncrementalMaintenance(void);
static void CloseMatViewIncrementalMaintenance(void);
+static void apply_delta(Oid matviewOid, Oid tempOid_new, Oid tempOid_old,
+ Query *query, Oid relowner, int save_sec_context);
+static SPIPlanPtr get_plan_for_recalc_min_max(Oid matviewOid, const char *min_max_list,
+ const char *group_keys, int nkeys, Oid *keyTypes);
+static SPIPlanPtr get_plan_for_set_min_max(char *matviewname, const char *min_max_list,
+ int num_min_max, Oid *valTypes);
+
+static Query *get_matview_query(Relation matviewRel);
+
+
/*
* SetMatViewPopulatedState
* Mark a materialized view as populated, or not.
@@ -114,6 +137,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 +203,6 @@ ExecRefreshMatView(RefreshMatViewStmt *stmt, const char *queryString,
{
Oid matviewOid;
Relation matviewRel;
- RewriteRule *rule;
- List *actions;
Query *dataQuery;
Oid tableSpace;
Oid relowner;
@@ -187,32 +248,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 +284,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 +342,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 +400,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 +437,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 +958,1024 @@ CloseMatViewIncrementalMaintenance(void)
matview_maintenance_depth--;
Assert(matview_maintenance_depth >= 0);
}
+
+/*
+ * IVM trigger function
+ */
+
+Datum
+IVM_immediate_maintenance(PG_FUNCTION_ARGS)
+{
+ TriggerData *trigdata = (TriggerData *) fcinfo->context;
+ Relation rel;
+ Oid relid;
+ Oid matviewOid;
+ Query *query, *old_delta_qry, *new_delta_qry;
+ 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;
+ char relpersistence;
+ Oid save_userid;
+ int save_sec_context;
+ int save_nestlevel;
+
+ ParseState *pstate;
+ QueryEnvironment *queryEnv = create_queryEnv();
+
+ Const *dmy_arg = makeConst(INT4OID,
+ -1,
+ InvalidOid,
+ sizeof(int32),
+ Int32GetDatum(1),
+ false,
+ true); /* pass by value */
+
+ /* Create a dummy ParseState for addRangeTableEntryForENR */
+ pstate = make_parsestate(NULL);
+ pstate->p_queryEnv = queryEnv;
+ pstate->p_expr_kind = EXPR_KIND_SELECT_TARGET;
+
+ names = stringToQualifiedNameList(matviewname);
+
+ /*
+ * 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);
+
+ matviewRel = table_open(matviewOid, NoLock);
+
+ /*
+ * 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());
+
+ /* Make sure it is a materialized view. */
+ if (matviewRel->rd_rel->relkind != RELKIND_MATVIEW)
+ ereport(ERROR,
+ (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+ errmsg("\"%s\" is not a materialized view",
+ RelationGetRelationName(matviewRel))));
+
+ rel = trigdata->tg_relation;
+ relid = rel->rd_id;
+
+ /* get view query*/
+ query = get_matview_query(matviewRel);
+
+ new_delta_qry = copyObject(query);
+ old_delta_qry = copyObject(query);
+
+ if (trigdata->tg_newtable)
+ {
+ RangeTblEntry *rte;
+ ListCell *lc;
+
+ TargetEntry *tle;
+ Node *node;
+ FuncCall *fn;
+
+ EphemeralNamedRelation enr =
+ palloc(sizeof(EphemeralNamedRelationData));
+
+ enr->md.name = trigdata->tg_trigger->tgnewtable;
+ enr->md.reliddesc = trigdata->tg_relation->rd_id;
+ enr->md.tupdesc = NULL;
+ enr->md.enrtype = ENR_NAMED_TUPLESTORE;
+ enr->md.enrtuples = tuplestore_tuple_count(trigdata->tg_newtable);
+ enr->reldata = trigdata->tg_newtable;
+ register_ENR(queryEnv, enr);
+
+ rte = addRangeTableEntryForENR(pstate, makeRangeVar(NULL, enr->md.name, -1), true);
+ new_delta_qry->rtable = lappend(new_delta_qry->rtable, rte);
+
+ foreach(lc, new_delta_qry->rtable)
+ {
+ RangeTblEntry *r = (RangeTblEntry*) lfirst(lc);
+ if (r->relid == relid)
+ {
+ lfirst(lc) = rte;
+ break;
+ }
+ }
+
+ if (query->hasAggs)
+ {
+ ListCell *lc;
+ List *agg_counts = NIL;
+ AttrNumber next_resno = list_length(query->targetList) + 1;
+ Node *node;
+
+ foreach(lc, query->targetList)
+ {
+ TargetEntry *tle = (TargetEntry *) lfirst(lc);
+ TargetEntry *tle_count;
+
+ 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)
+ {
+ 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, 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++;
+ }
+ }
+
+ }
+ new_delta_qry->targetList = list_concat(new_delta_qry->targetList, agg_counts);
+ }
+
+ fn = makeFuncCall(list_make1(makeString("count")), NIL, -1);
+ fn->agg_star = true;
+ if (!new_delta_qry->groupClause && !new_delta_qry->hasAggs)
+ new_delta_qry->groupClause = transformDistinctClause(NULL, &new_delta_qry->targetList, new_delta_qry->sortClause, false);
+
+ node = ParseFuncOrColumn(pstate, fn->funcname, NIL, NULL, fn, false, -1);
+
+ tle = makeTargetEntry((Expr *) node,
+ list_length(new_delta_qry->targetList) + 1,
+ NULL,
+ false);
+ new_delta_qry->targetList = lappend(new_delta_qry->targetList, tle);
+ new_delta_qry->hasAggs = true;
+ }
+
+ if (trigdata->tg_oldtable)
+ {
+ RangeTblEntry *rte;
+ ListCell *lc;
+
+ TargetEntry *tle;
+ Node *node;
+ FuncCall *fn = makeFuncCall(list_make1(makeString("count")), NIL, -1);
+
+ EphemeralNamedRelation enr =
+ palloc(sizeof(EphemeralNamedRelationData));
+
+ enr->md.name = trigdata->tg_trigger->tgoldtable;
+ enr->md.reliddesc = trigdata->tg_relation->rd_id;
+ enr->md.tupdesc = NULL;
+ enr->md.enrtype = ENR_NAMED_TUPLESTORE;
+ enr->md.enrtuples = tuplestore_tuple_count(trigdata->tg_oldtable);
+ enr->reldata = trigdata->tg_oldtable;
+ register_ENR(queryEnv, enr);
+
+ rte = addRangeTableEntryForENR(pstate, makeRangeVar(NULL, enr->md.name, -1), true);
+ old_delta_qry->rtable = lappend(old_delta_qry->rtable, rte);
+
+ foreach(lc, old_delta_qry->rtable)
+ {
+ RangeTblEntry *r = (RangeTblEntry*) lfirst(lc);
+ if (r->relid == relid)
+ {
+ lfirst(lc) = rte;
+ break;
+ }
+ }
+
+ if (query->hasAggs)
+ {
+ ListCell *lc;
+ List *agg_counts = NIL;
+ AttrNumber next_resno = list_length(query->targetList) + 1;
+ Node *node;
+
+ foreach(lc, query->targetList)
+ {
+ TargetEntry *tle = (TargetEntry *) lfirst(lc);
+ TargetEntry *tle_count;
+
+ 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)
+ {
+ 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, 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++;
+ }
+ }
+
+ }
+ old_delta_qry->targetList = list_concat(old_delta_qry->targetList, agg_counts);
+ }
+
+ fn = makeFuncCall(list_make1(makeString("count")), NIL, -1);
+ fn->agg_star = true;
+
+ if (!old_delta_qry->groupClause && !old_delta_qry->hasAggs)
+ old_delta_qry->groupClause = transformDistinctClause(NULL, &old_delta_qry->targetList, old_delta_qry->sortClause, false);
+
+ node = ParseFuncOrColumn(pstate, fn->funcname, NIL, NULL, fn, false, -1);
+ tle = makeTargetEntry((Expr *) node,
+ list_length(old_delta_qry->targetList) + 1,
+ NULL,
+ false);
+ old_delta_qry->targetList = lappend(old_delta_qry->targetList, tle);
+ old_delta_qry->hasAggs = true;
+ }
+
+
+ /*
+ * 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.
+ */
+ CheckTableNotInUse(matviewRel, "REFRESH MATERIALIZED VIEW");
+
+ 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();
+
+ tableSpace = GetDefaultTablespace(RELPERSISTENCE_TEMP, false);
+ relpersistence = RELPERSISTENCE_TEMP;
+
+ /*
+ * Create the transient table that will receive the regenerated data. Lock
+ * it against access by any other process until commit (by which time it
+ * will be gone).
+ */
+ if (trigdata->tg_newtable)
+ {
+ OIDDelta_new = make_new_heap(matviewOid, tableSpace, relpersistence,
+ ExclusiveLock);
+ LockRelationOid(OIDDelta_new, AccessExclusiveLock);
+ dest_new = CreateTransientRelDestReceiver(OIDDelta_new);
+ }
+ if (trigdata->tg_oldtable)
+ {
+ if (trigdata->tg_newtable)
+ OIDDelta_old = make_new_heap(OIDDelta_new, tableSpace, relpersistence,
+ ExclusiveLock);
+ else
+ OIDDelta_old = make_new_heap(matviewOid, tableSpace, relpersistence,
+ ExclusiveLock);
+ LockRelationOid(OIDDelta_old, AccessExclusiveLock);
+ dest_old = CreateTransientRelDestReceiver(OIDDelta_old);
+ }
+
+ /*
+ * Now lock down security-restricted operations.
+ */
+ SetUserIdAndSecContext(relowner,
+ save_sec_context | SECURITY_RESTRICTED_OPERATION);
+
+ /* Generate the data. */
+ if (trigdata->tg_newtable)
+ refresh_matview_datafill(dest_new, new_delta_qry, queryEnv, NULL);
+ if (trigdata->tg_oldtable)
+ refresh_matview_datafill(dest_old, old_delta_qry, queryEnv, NULL);
+
+ PG_TRY();
+ {
+ apply_delta(matviewOid, OIDDelta_new, OIDDelta_old,
+ query, relowner, save_sec_context);
+ }
+ PG_CATCH();
+ {
+ matview_maintenance_depth = old_depth;
+ PG_RE_THROW();
+ }
+ PG_END_TRY();
+
+ /* 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);
+
+ return PointerGetDatum(NULL);
+}
+
+static void
+apply_delta(Oid matviewOid, Oid tempOid_new, Oid tempOid_old,
+ Query *query, Oid relowner, int save_sec_context)
+{
+ 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"))
+ {
+ appendStringInfo(&update_aggs_old,
+ "%s = %s - %s",
+ quote_qualified_identifier(NULL, resname),
+ quote_qualified_identifier("mv", resname),
+ quote_qualified_identifier("t", resname)
+ );
+ appendStringInfo(&update_aggs_new,
+ "%s = %s + %s",
+ quote_qualified_identifier(NULL, resname),
+ quote_qualified_identifier("mv", resname),
+ quote_qualified_identifier("diff", resname)
+ );
+
+ appendStringInfo(&diff_aggs_buf, "%s",
+ quote_qualified_identifier("diff", resname)
+ );
+ }
+ else if (!strcmp(aggname, "sum"))
+ {
+ appendStringInfo(&update_aggs_old,
+ "%s = CASE WHEN %s = %s THEN NULL ELSE "
+ "COALESCE(%s,0) - COALESCE(%s, 0) END, "
+ "%s = %s - %s",
+ quote_qualified_identifier(NULL, resname),
+ quote_qualified_identifier("mv", makeObjectName("__ivm_count",resname,"_")),
+ quote_qualified_identifier("t", makeObjectName("__ivm_count",resname,"_")),
+
+ quote_qualified_identifier("mv", resname),
+ quote_qualified_identifier("t", resname),
+
+ quote_qualified_identifier(NULL, makeObjectName("__ivm_count",resname,"_")),
+ quote_qualified_identifier("mv", makeObjectName("__ivm_count",resname,"_")),
+ quote_qualified_identifier("t", makeObjectName("__ivm_count",resname,"_"))
+ );
+ appendStringInfo(&update_aggs_new,
+ "%s = CASE WHEN %s = 0 AND %s = 0 THEN NULL ELSE "
+ "COALESCE(%s,0) + COALESCE(%s, 0) END, "
+ "%s = %s + %s",
+ quote_qualified_identifier(NULL, resname),
+ quote_qualified_identifier("mv", makeObjectName("__ivm_count",resname,"_")),
+ quote_qualified_identifier("diff", makeObjectName("__ivm_count",resname,"_")),
+
+ quote_qualified_identifier("mv", resname),
+ quote_qualified_identifier("diff", resname),
+
+ quote_qualified_identifier(NULL, makeObjectName("__ivm_count",resname,"_")),
+ quote_qualified_identifier("mv", makeObjectName("__ivm_count",resname,"_")),
+ quote_qualified_identifier("diff", makeObjectName("__ivm_count",resname,"_"))
+ );
+
+ appendStringInfo(&diff_aggs_buf, "%s, %s",
+ quote_qualified_identifier("diff", resname),
+ quote_qualified_identifier("diff", makeObjectName("__ivm_count",resname,"_"))
+ );
+ }
+ else if (!strcmp(aggname, "avg"))
+ {
+ appendStringInfo(&update_aggs_old,
+ "%s = CASE WHEN %s = %s THEN NULL ELSE "
+ "(COALESCE(%s,0) - COALESCE(%s, 0))::%s / (%s - %s) END, "
+ "%s = COALESCE(%s,0) - COALESCE(%s, 0), "
+ "%s = %s - %s",
+
+ quote_qualified_identifier(NULL, resname),
+ quote_qualified_identifier("mv", makeObjectName("__ivm_count",resname,"_")),
+ quote_qualified_identifier("t", makeObjectName("__ivm_count",resname,"_")),
+
+ quote_qualified_identifier("mv", makeObjectName("__ivm_sum",resname,"_")),
+ quote_qualified_identifier("t", makeObjectName("__ivm_sum",resname,"_")),
+ aggtype,
+ quote_qualified_identifier("mv", makeObjectName("__ivm_count",resname,"_")),
+ quote_qualified_identifier("t", makeObjectName("__ivm_count",resname,"_")),
+
+ quote_qualified_identifier(NULL, makeObjectName("__ivm_sum",resname,"_")),
+ quote_qualified_identifier("mv", makeObjectName("__ivm_sum",resname,"_")),
+ quote_qualified_identifier("t", makeObjectName("__ivm_sum",resname,"_")),
+
+ quote_qualified_identifier(NULL, makeObjectName("__ivm_count",resname,"_")),
+ quote_qualified_identifier("mv", makeObjectName("__ivm_count",resname,"_")),
+ quote_qualified_identifier("t", makeObjectName("__ivm_count",resname,"_"))
+ );
+ appendStringInfo(&update_aggs_new,
+ "%s = CASE WHEN %s = 0 AND %s = 0 THEN NULL ELSE "
+ "(COALESCE(%s,0)+ COALESCE(%s, 0))::%s / (%s + %s) END, "
+ "%s = COALESCE(%s,0) + COALESCE(%s, 0), "
+ "%s = %s + %s",
+
+ quote_qualified_identifier(NULL, resname),
+ quote_qualified_identifier("mv", makeObjectName("__ivm_count",resname,"_")),
+ quote_qualified_identifier("diff", makeObjectName("__ivm_count",resname,"_")),
+
+ quote_qualified_identifier("mv", makeObjectName("__ivm_sum",resname,"_")),
+ quote_qualified_identifier("diff", makeObjectName("__ivm_sum",resname,"_")),
+ aggtype,
+ quote_qualified_identifier("mv", makeObjectName("__ivm_count",resname,"_")),
+ quote_qualified_identifier("diff", makeObjectName("__ivm_count",resname,"_")),
+
+ quote_qualified_identifier(NULL, makeObjectName("__ivm_sum",resname,"_")),
+ quote_qualified_identifier("mv", makeObjectName("__ivm_sum",resname,"_")),
+ quote_qualified_identifier("diff", makeObjectName("__ivm_sum",resname,"_")),
+
+ quote_qualified_identifier(NULL, makeObjectName("__ivm_count",resname,"_")),
+ quote_qualified_identifier("mv", makeObjectName("__ivm_count",resname,"_")),
+ quote_qualified_identifier("diff", makeObjectName("__ivm_count",resname,"_"))
+ );
+
+ 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");
+
+ 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));
+
+ appendStringInfo(&update_aggs_old,
+ "%s = CASE WHEN %s = %s THEN NULL ELSE "
+ "%s END, "
+ "%s = %s - %s",
+ quote_qualified_identifier(NULL, resname),
+ quote_qualified_identifier("mv", makeObjectName("__ivm_count",resname,"_")),
+ quote_qualified_identifier("t", makeObjectName("__ivm_count",resname,"_")),
+
+ quote_qualified_identifier("mv", resname),
+
+ quote_qualified_identifier(NULL, makeObjectName("__ivm_count",resname,"_")),
+ quote_qualified_identifier("mv", makeObjectName("__ivm_count",resname,"_")),
+ quote_qualified_identifier("t", makeObjectName("__ivm_count",resname,"_"))
+ );
+ appendStringInfo(&update_aggs_new,
+ "%s = CASE WHEN %s = 0 AND %s = 0 THEN NULL ELSE "
+ "%s(%s,%s) END, "
+ "%s = %s + %s",
+ quote_qualified_identifier(NULL, resname),
+ quote_qualified_identifier("mv", makeObjectName("__ivm_count",resname,"_")),
+ quote_qualified_identifier("diff", makeObjectName("__ivm_count",resname,"_")),
+
+ is_min ? "least" : "greatest",
+ quote_qualified_identifier("mv", resname),
+ quote_qualified_identifier("diff", resname),
+
+ quote_qualified_identifier(NULL, makeObjectName("__ivm_count",resname,"_")),
+ quote_qualified_identifier("mv", makeObjectName("__ivm_count",resname,"_")),
+ quote_qualified_identifier("diff", makeObjectName("__ivm_count",resname,"_"))
+ );
+
+ 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);
+ }
+
+ SetUserIdAndSecContext(relowner,
+ save_sec_context | SECURITY_LOCAL_USERID_CHANGE);
+
+ 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 ? 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, *minmaxTypes;
+ char *keyNulls, *minmaxNulls;
+ Datum *keyVals, *minmaxVals;
+
+ keyTypes = palloc(sizeof(Oid) * num_group_keys);
+ keyNulls = palloc(sizeof(char) * num_group_keys);
+ keyVals = palloc(sizeof(Datum) * num_group_keys);
+
+ 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));
+
+ Assert(tupdesc_recalc->natts == num_group_keys + 1);
+
+ for (i = 0; i < num_group_keys; i++)
+ keyTypes[i] = TupleDescAttr(tupdesc_recalc, i+1)->atttypid;
+
+ for (i=0; i< processed; i++)
+ {
+ int j;
+ bool isnull;
+ SPIPlanPtr plan;
+ SPITupleTable *tuptable_minmax;
+ TupleDesc tupdesc_minmax;
+
+
+ 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);
+
+ if (SPI_execute_plan(plan, keyVals, keyNulls, false, 0) != SPI_OK_SELECT)
+ elog(ERROR, "SPI_execcute_plan");
+ 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(matviewname, min_or_max_buf.data,
+ num_min_or_max, minmaxTypes);
+
+ if (SPI_execute_plan(plan, minmaxVals, minmaxNulls, false, 0) != SPI_OK_UPDATE)
+ elog(ERROR, "SPI_execcute_plan");
+
+ }
+ }
+
+ }
+ 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);
+
+ /* Clean up temp tables. */
+ 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);
+ }
+ 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);
+ }
+
+ /* Close SPI context. */
+ if (SPI_finish() != SPI_OK_FINISH)
+ elog(ERROR, "SPI_finish failed");
+}
+
+
+static SPIPlanPtr
+get_plan_for_recalc_min_max(Oid matviewOid, const char *min_max_list,
+ const char *group_keys, int nkeys, Oid *keyTypes)
+{
+ StringInfoData str;
+ char *viewdef;
+ int i;
+
+ if (plan_for_recalc_min_max)
+ return plan_for_recalc_min_max;
+
+ /* 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 WHERE (%s) = (",
+ min_max_list, viewdef, group_keys);
+
+ for (i = 1; i <= nkeys; i++)
+ appendStringInfo(&str, "%s$%d", (i==1 ? "" : ", "),i );
+
+ appendStringInfo(&str, ")");
+
+ plan_for_recalc_min_max = SPI_prepare(str.data, nkeys, keyTypes);
+ SPI_keepplan(plan_for_recalc_min_max);
+
+ return plan_for_recalc_min_max;
+}
+
+static SPIPlanPtr
+get_plan_for_set_min_max(char *matviewname, const char *min_max_list,
+ int num_min_max, Oid *valTypes)
+{
+ StringInfoData str;
+ int i;
+
+ if (plan_for_set_min_max)
+ return plan_for_set_min_max;
+
+ 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, ") WHERE ctid = $%d", i);
+
+ plan_for_set_min_max = SPI_prepare(str.data, num_min_max + 1, valTypes);
+ SPI_keepplan(plan_for_set_min_max);
+
+ return plan_for_set_min_max;
+}
+
+
+/*
+ * 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);
+}
diff --git a/src/backend/nodes/copyfuncs.c b/src/backend/nodes/copyfuncs.c
index 6414aded0e..4f54258005 100644
--- a/src/backend/nodes/copyfuncs.c
+++ b/src/backend/nodes/copyfuncs.c
@@ -2361,6 +2361,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 4f2ebe5118..608d477bd5 100644
--- a/src/backend/nodes/equalfuncs.c
+++ b/src/backend/nodes/equalfuncs.c
@@ -2641,6 +2641,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 86c31a48c9..be07641e82 100644
--- a/src/backend/nodes/outfuncs.c
+++ b/src/backend/nodes/outfuncs.c
@@ -3044,6 +3044,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 6c2626ee62..aa01e205c3 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 c97bb367f8..29040126e0 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
@@ -4054,30 +4055,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;
}
;
@@ -4094,9 +4097,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; }
;
@@ -15128,6 +15136,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 c96ef8595a..5f2cb21b3f 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 c13c08a97b..296cc53ffd 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 248860758c..531da6f879 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 774cc764ff..cc7388d412 100644
--- a/src/bin/psql/describe.c
+++ b/src/bin/psql/describe.c
@@ -1491,6 +1491,7 @@ describeOneTableDetails(const char *schemaname,
char relpersistence;
char relreplident;
char *relam;
+ bool isivm;
} tableinfo;
bool show_column_details = false;
@@ -1511,6 +1512,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"
@@ -1687,6 +1689,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;
@@ -3291,6 +3297,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 3f7001fb69..c481b3a2e6 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},
@@ -2486,7 +2487,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 (");
@@ -2734,13 +2735,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 0902dce5f1..6e1057c5b6 100644
--- a/src/include/catalog/pg_proc.dat
+++ b/src/include/catalog/pg_proc.dat
@@ -10701,4 +10701,9 @@
proname => 'pg_partition_root', prorettype => 'regclass',
proargtypes => 'regclass', prosrc => 'pg_partition_root' },
+# IVM
+{ oid => '786', descr => 'ivm trigger',
+ 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..8dd8193d9c 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,6 @@ extern DestReceiver *CreateTransientRelDestReceiver(Oid oid);
extern bool MatViewIncrementalMaintenanceIsEnabled(void);
+extern Datum IVM_immediate_maintenance(PG_FUNCTION_ARGS);
+
#endif /* MATVIEW_H */
diff --git a/src/include/nodes/parsenodes.h b/src/include/nodes/parsenodes.h
index 94ded3c135..5c8a5ae3ca 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 b0fe19ebc5..9da0a4a9f9 100644
--- a/src/include/utils/rel.h
+++ b/src/include/utils/rel.h
@@ -565,6 +565,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..68060cc179
--- /dev/null
+++ b/src/test/regress/expected/incremental_matview.out
@@ -0,0 +1,306 @@
+-- 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;
+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;
+-- restriction of incremental view maintenance
+-- 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 8fb55f045e..39e79d8d10 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
+test: create_table_like alter_generic alter_operator misc async dbsize misc_functions sysviews tsrf tidscan 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 a39ca1012a..552c80c851 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..2dd968f709
--- /dev/null
+++ b/src/test/regress/sql/incremental_matview.sql
@@ -0,0 +1,117 @@
+-- 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;
+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;
+
+-- restriction of incremental view maintenance
+
+-- 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;
