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$<iteration count>:&l
True if table or index is a partition
+
+ relisivm
+ bool
+
+ True if materialized view enables incremental view maintenance
+
+
relrewrite
oid
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
-CREATE MATERIALIZED VIEW [ IF NOT EXISTS ] table_name
+CREATE [ INCREMENTAL ] MATERIALIZED VIEW [ IF NOT EXISTS ] table_name
[ (column_name [, ...] ) ]
[ USING method ]
[ WITH ( storage_parameter [= value] [, ... ] ) ]
@@ -54,6 +54,16 @@ CREATE MATERIALIZED VIEW [ IF NOT EXISTS ] table_name
Parameters
+
+ INCREMENTAL
+
+
+ If specified, a materialized view enables incremental view maintenance.
+ You can replace only the contents of a materialized view, which based rows are changed.
+
+
+
+
IF NOT EXISTS
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 OptTempTableName
%type into_clause create_as_target create_mv_target
+%type incremental
%type createfunc_opt_item common_func_opt_item dostmt_opt_item
%type 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 with AS */
- else if (Matches("CREATE", "MATERIALIZED", "VIEW", MatchAny))
+ /* Complete CREATE MATERIALIZED VIEW with AS */
+ else if (Matches("CREATE", "MATERIALIZED", "VIEW", MatchAny) ||
+ Matches("CREATE", "INCREMENTAL", "MATERIALIZED", "VIEW", MatchAny))
COMPLETE_WITH("AS");
/* Complete "CREATE MATERIALIZED VIEW 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;