From 4739745e3ff97ed6b1b6af0516e7ce02d83ca0e1 Mon Sep 17 00:00:00 2001 From: Yugo Nagata Date: Fri, 29 May 2026 18:06:31 +0900 Subject: [PATCH v37 06/11] Add Incremental View Maintenance support In this implementation, AFTER triggers are used to collect tuplestores containing transition table contents. When multiple tables are changed, multiple AFTER triggers are invoked, then the final AFTER trigger performs actual update of the matview. In addition, BEFORE triggers are also used to handle global information for view maintenance. To calculate view deltas, we need both pre-state and post-state of base tables. Post-update states are available in AFTER trigger, and pre-update states can be calculated by removing inserted tuples and appending deleted tuples. Insterted tuples are filtered using the snapshot taken before table modification, and deleted tuples are contained in the old transition table. Incrementally Maintainable Materialized Views (IMMV) can contain duplicated tuples. This patch also allows self-join, simultaneous updates of more than one base table, and multiple updates of the same base table. --- src/backend/access/transam/xact.c | 8 + src/backend/commands/createas.c | 709 +++++++ src/backend/commands/matview.c | 1835 ++++++++++++++++- src/backend/commands/tablecmds.c | 4 + .../utils/activity/wait_event_names.txt | 1 + src/include/catalog/pg_proc.dat | 10 + src/include/commands/createas.h | 4 + src/include/commands/matview.h | 11 + src/include/storage/lwlocklist.h | 1 + src/include/storage/subsystemlist.h | 3 + 10 files changed, 2548 insertions(+), 38 deletions(-) diff --git a/src/backend/access/transam/xact.c b/src/backend/access/transam/xact.c index 5586fbe5b07..302a4822d50 100644 --- a/src/backend/access/transam/xact.c +++ b/src/backend/access/transam/xact.c @@ -37,6 +37,7 @@ #include "catalog/pg_enum.h" #include "catalog/storage.h" #include "commands/async.h" +#include "commands/matview.h" #include "commands/tablecmds.h" #include "commands/trigger.h" #include "common/pg_prng.h" @@ -2322,6 +2323,9 @@ CommitTransaction(void) CallXactCallbacks(is_parallel_worker ? XACT_EVENT_PARALLEL_PRE_COMMIT : XACT_EVENT_PRE_COMMIT); + /* Store the transaction ID that updated the view incrementally */ + AtPreCommit_IVM(); + /* * If this xact has started any unfinished parallel operation, clean up * its workers, warning about leaked resources. (But we don't actually @@ -2971,6 +2975,7 @@ AbortTransaction(void) AtAbort_Notify(); AtEOXact_RelationMap(false, is_parallel_worker); AtAbort_Twophase(); + AtAbort_IVM(InvalidSubTransactionId); /* * Advertise the fact that we aborted in pg_xact (assuming that we got as @@ -5301,6 +5306,9 @@ AbortSubTransaction(void) UnlockBuffers(); + /* Clean up hash entries for incremental view maintenance */ + AtAbort_IVM(s->subTransactionId); + /* Reset WAL record construction state */ XLogResetInsertion(); diff --git a/src/backend/commands/createas.c b/src/backend/commands/createas.c index 6dbb831ca89..a499688b79d 100644 --- a/src/backend/commands/createas.c +++ b/src/backend/commands/createas.c @@ -29,18 +29,32 @@ #include "access/tableam.h" #include "access/xact.h" #include "catalog/namespace.h" +#include "catalog/index.h" +#include "catalog/pg_am_d.h" +#include "catalog/pg_constraint.h" +#include "catalog/pg_inherits.h" +#include "catalog/pg_trigger.h" #include "catalog/toasting.h" #include "commands/createas.h" +#include "commands/defrem.h" #include "commands/matview.h" #include "commands/prepare.h" #include "commands/tablecmds.h" +#include "commands/tablespace.h" +#include "commands/trigger.h" #include "commands/view.h" #include "executor/execdesc.h" #include "executor/executor.h" +#include "miscadmin.h" #include "nodes/makefuncs.h" #include "nodes/nodeFuncs.h" #include "nodes/queryjumble.h" +#include "optimizer/optimizer.h" +#include "optimizer/prep.h" #include "parser/analyze.h" +#include "parser/parser.h" +#include "parser/parsetree.h" +#include "parser/parse_clause.h" #include "rewrite/rewriteHandler.h" #include "tcop/tcopprot.h" #include "utils/builtins.h" @@ -70,6 +84,12 @@ static bool intorel_receive(TupleTableSlot *slot, DestReceiver *self); static void intorel_shutdown(DestReceiver *self); static void intorel_destroy(DestReceiver *self); +static void CreateIvmTriggersOnBaseTablesRecurse(Query *qry, Node *node, Oid matviewOid, + List **relids, bool ex_lock); +static void CreateIvmTrigger(Oid relOid, Oid viewOid, int16 type, int16 timing, bool ex_lock); +static void check_ivm_restriction(Node *node); +static bool check_ivm_restriction_walker(Node *node, void *context); +static Bitmapset *get_primary_key_attnos_from_query(Query *query, List **constraintList); /* * create_ctas_internal @@ -277,6 +297,18 @@ ExecCreateTableAs(ParseState *pstate, CreateTableAsStmt *stmt, into->skipData = true; } + if (is_matview && into->ivm) + { + /* check if the query is supported in IMMV definition */ + if (contain_mutable_functions((Node *) query)) + ereport(ERROR, + (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg("mutable function is not supported on incrementally maintainable materialized view"), + errhint("functions must be marked IMMUTABLE"))); + + check_ivm_restriction((Node *) query); + } + if (into->skipData) { /* @@ -287,15 +319,36 @@ ExecCreateTableAs(ParseState *pstate, CreateTableAsStmt *stmt, */ address = create_ctas_nodata(query->targetList, into); + /* + * For IVM, mark the matview before refresh so RelationIsIVM() + * returns true inside RefreshMatViewByOid. + */ + if (is_matview && into->ivm) + { + Relation matviewRel = table_open(address.objectId, NoLock); + + SetMatViewIVMState(matviewRel, true); + table_close(matviewRel, NoLock); + CommandCounterIncrement(); + } + /* * For materialized views, reuse the REFRESH logic, which locks down * security-restricted operations and restricts the search_path. This * reduces the chance that a subsequent refresh will fail. */ if (do_refresh) + { RefreshMatViewByOid(address.objectId, true, false, false, pstate->p_sourcetext, qc); + if (is_matview && into->ivm && IsolationUsesXactSnapshot()) + ereport(WARNING, + (errmsg("inconsistent view can be created in isolation level SERIALIZABLE or REPEATABLE READ"), + errdetail("The view may not include effects of a concurrent transaction."), + errhint("The view with incremental maintenance should be created in isolation level READ COMMITTED, " + "or refreshed manually to make sure the view is consistent."))); + } } else { @@ -635,3 +688,659 @@ intorel_destroy(DestReceiver *self) { pfree(self); } + +/* + * CreateIvmTriggersOnBaseTables -- create IVM triggers on all base tables + */ +void +CreateIvmTriggersOnBaseTables(Query *qry, Oid matviewOid) +{ + List *relids = NIL; + bool ex_lock = false; + RangeTblEntry *rte; + + /* Immediately return if we don't have any base tables. */ + if (list_length(qry->rtable) < 1) + return; + + /* + * If the view has more than one base tables, we need an exclusive lock + * on the view so that the view would be maintained serially to avoid + * the inconsistency that occurs when two base tables are modified in + * concurrent transactions. However, if the view has only one table, + * we can use a weaker lock. + * + * The type of lock should be determined here, because if we check the + * view definition at maintenance time, we need to acquire a weaker lock, + * and upgrading the lock level after this increases probability of + * deadlock. + */ + + rte = list_nth(qry->rtable, 0); + if (list_length(qry->rtable) > 1 || rte->rtekind != RTE_RELATION) + ex_lock = true; + + CreateIvmTriggersOnBaseTablesRecurse(qry, (Node *)qry, matviewOid, &relids, ex_lock); + + list_free(relids); +} + +static void +CreateIvmTriggersOnBaseTablesRecurse(Query *qry, Node *node, Oid matviewOid, + List **relids, bool ex_lock) +{ + if (node == NULL) + return; + + /* This can recurse, so check for excessive recursion */ + check_stack_depth(); + + switch (nodeTag(node)) + { + case T_Query: + { + Query *query = (Query *) node; + + CreateIvmTriggersOnBaseTablesRecurse(qry, (Node *)query->jointree, matviewOid, relids, ex_lock); + } + break; + + case T_RangeTblRef: + { + int rti = ((RangeTblRef *) node)->rtindex; + RangeTblEntry *rte = rt_fetch(rti, qry->rtable); + + if (rte->rtekind == RTE_RELATION && !list_member_oid(*relids, rte->relid)) + { + CreateIvmTrigger(rte->relid, matviewOid, TRIGGER_TYPE_INSERT, TRIGGER_TYPE_BEFORE, ex_lock); + CreateIvmTrigger(rte->relid, matviewOid, TRIGGER_TYPE_DELETE, TRIGGER_TYPE_BEFORE, ex_lock); + CreateIvmTrigger(rte->relid, matviewOid, TRIGGER_TYPE_UPDATE, TRIGGER_TYPE_BEFORE, ex_lock); + CreateIvmTrigger(rte->relid, matviewOid, TRIGGER_TYPE_TRUNCATE, TRIGGER_TYPE_BEFORE, true); + CreateIvmTrigger(rte->relid, matviewOid, TRIGGER_TYPE_INSERT, TRIGGER_TYPE_AFTER, ex_lock); + CreateIvmTrigger(rte->relid, matviewOid, TRIGGER_TYPE_DELETE, TRIGGER_TYPE_AFTER, ex_lock); + CreateIvmTrigger(rte->relid, matviewOid, TRIGGER_TYPE_UPDATE, TRIGGER_TYPE_AFTER, ex_lock); + CreateIvmTrigger(rte->relid, matviewOid, TRIGGER_TYPE_TRUNCATE, TRIGGER_TYPE_AFTER, true); + + *relids = lappend_oid(*relids, rte->relid); + } + } + break; + + case T_FromExpr: + { + FromExpr *f = (FromExpr *) node; + ListCell *l; + + foreach(l, f->fromlist) + CreateIvmTriggersOnBaseTablesRecurse(qry, lfirst(l), matviewOid, relids, ex_lock); + } + break; + + case T_JoinExpr: + { + JoinExpr *j = (JoinExpr *) node; + + CreateIvmTriggersOnBaseTablesRecurse(qry, j->larg, matviewOid, relids, ex_lock); + CreateIvmTriggersOnBaseTablesRecurse(qry, j->rarg, matviewOid, relids, ex_lock); + } + break; + + default: + elog(ERROR, "unrecognized node type: %d", (int) nodeTag(node)); + } +} + +/* + * CreateIvmTrigger -- create IVM trigger on a base table + */ +static void +CreateIvmTrigger(Oid relOid, Oid viewOid, int16 type, int16 timing, bool ex_lock) +{ + ObjectAddress refaddr; + ObjectAddress address; + CreateTrigStmt *ivm_trigger; + List *transitionRels = NIL; + + Assert(timing == TRIGGER_TYPE_BEFORE || timing == TRIGGER_TYPE_AFTER); + + refaddr.classId = RelationRelationId; + refaddr.objectId = viewOid; + refaddr.objectSubId = 0; + + ivm_trigger = makeNode(CreateTrigStmt); + ivm_trigger->relation = NULL; + ivm_trigger->row = false; + + ivm_trigger->timing = timing; + ivm_trigger->events = type; + + switch (type) + { + case TRIGGER_TYPE_INSERT: + ivm_trigger->trigname = (timing == TRIGGER_TYPE_BEFORE ? "IVM_trigger_ins_before" : "IVM_trigger_ins_after"); + break; + case TRIGGER_TYPE_DELETE: + ivm_trigger->trigname = (timing == TRIGGER_TYPE_BEFORE ? "IVM_trigger_del_before" : "IVM_trigger_del_after"); + break; + case TRIGGER_TYPE_UPDATE: + ivm_trigger->trigname = (timing == TRIGGER_TYPE_BEFORE ? "IVM_trigger_upd_before" : "IVM_trigger_upd_after"); + break; + case TRIGGER_TYPE_TRUNCATE: + ivm_trigger->trigname = (timing == TRIGGER_TYPE_BEFORE ? "IVM_trigger_truncate_before" : "IVM_trigger_truncate_after"); + break; + default: + elog(ERROR, "unsupported trigger type"); + } + + if (timing == TRIGGER_TYPE_AFTER) + { + if (type == TRIGGER_TYPE_INSERT || type == TRIGGER_TYPE_UPDATE) + { + TriggerTransition *n = makeNode(TriggerTransition); + n->name = "__ivm_newtable"; + n->isNew = true; + n->isTable = true; + + transitionRels = lappend(transitionRels, n); + } + if (type == TRIGGER_TYPE_DELETE || type == TRIGGER_TYPE_UPDATE) + { + TriggerTransition *n = makeNode(TriggerTransition); + n->name = "__ivm_oldtable"; + n->isNew = false; + n->isTable = true; + + transitionRels = lappend(transitionRels, n); + } + } + + /* + * XXX: When using DELETE or UPDATE, we must use exclusive lock for now + * because apply_old_delta(_with_count) uses ctid to identify the tuple + * to be deleted/deleted, but doesn't work in concurrent situations. + * + * If the view doesn't have aggregate, distinct, or tuple duplicate, + * then it would work even in concurrent situations. However, we don't have + * any way to guarantee the view has a unique key before opening the IMMV + * at the maintenance time because users may drop the unique index. + */ + + if (type == TRIGGER_TYPE_DELETE || type == TRIGGER_TYPE_UPDATE) + ex_lock = true; + + ivm_trigger->funcname = + (timing == TRIGGER_TYPE_BEFORE ? SystemFuncName("IVM_immediate_before") : SystemFuncName("IVM_immediate_maintenance")); + + ivm_trigger->columns = NIL; + ivm_trigger->transitionRels = transitionRels; + ivm_trigger->whenClause = NULL; + ivm_trigger->isconstraint = false; + ivm_trigger->deferrable = false; + ivm_trigger->initdeferred = false; + ivm_trigger->constrrel = NULL; + ivm_trigger->args = list_make2( + makeString(DatumGetPointer(DirectFunctionCall1(oidout, ObjectIdGetDatum(viewOid)))), + makeString(DatumGetPointer(DirectFunctionCall1(boolout, BoolGetDatum(ex_lock)))) + ); + + 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 --- look for specify nodes in the query tree + */ +static void +check_ivm_restriction(Node *node) +{ + check_ivm_restriction_walker(node, NULL); +} + +static bool +check_ivm_restriction_walker(Node *node, void *context) +{ + if (node == NULL) + return false; + + /* + * We currently don't support Sub-Query. + */ + if (IsA(node, SubPlan) || IsA(node, SubLink)) + ereport(ERROR, + (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg("subquery is not supported on incrementally maintainable materialized view"))); + + /* This can recurse, so check for excessive recursion */ + check_stack_depth(); + + switch (nodeTag(node)) + { + case T_Query: + { + Query *qry = (Query *)node; + ListCell *lc; + List *vars; + + /* if contained CTE, return error */ + if (qry->cteList != NIL) + ereport(ERROR, + (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg("CTE is not supported on incrementally maintainable materialized view"))); + if (qry->havingQual != NULL) + ereport(ERROR, + (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg(" HAVING clause is not supported on incrementally maintainable materialized view"))); + if (qry->sortClause != NIL) /* There is a possibility that we don't need to return an error */ + ereport(ERROR, + (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg("ORDER BY clause is not supported on incrementally maintainable materialized view"))); + if (qry->limitOffset != NULL || qry->limitCount != NULL) + ereport(ERROR, + (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg("LIMIT/OFFSET clause is not supported on incrementally maintainable materialized view"))); + if (qry->distinctClause) + ereport(ERROR, + (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg("DISTINCT is not supported on incrementally maintainable materialized view"))); + if (qry->hasDistinctOn) + ereport(ERROR, + (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg("DISTINCT ON is not supported on incrementally maintainable materialized view"))); + if (qry->hasWindowFuncs) + ereport(ERROR, + (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg("window functions are not supported on incrementally maintainable materialized view"))); + if (qry->groupingSets != NIL) + ereport(ERROR, + (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg("GROUPING SETS, ROLLUP, or CUBE clauses is not supported on incrementally maintainable materialized view"))); + if (qry->setOperations != NULL) + ereport(ERROR, + (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg("UNION/INTERSECT/EXCEPT statements are not supported on incrementally maintainable materialized view"))); + if (list_length(qry->targetList) == 0) + ereport(ERROR, + (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg("empty target list is not supported on incrementally maintainable materialized view"))); + if (qry->rowMarks != NIL) + ereport(ERROR, + (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg("FOR UPDATE/SHARE clause is not supported on incrementally maintainable materialized view"))); + + /* system column restrictions */ + vars = pull_vars_of_level((Node *) qry, 0); + foreach(lc, vars) + { + if (IsA(lfirst(lc), Var)) + { + Var *var = (Var *) lfirst(lc); + /* if system column, return error */ + if (var->varattno < 0) + ereport(ERROR, + (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg("system column is not supported on incrementally maintainable materialized view"))); + } + } + + /* check if type in the target list had an equality operator */ + foreach(lc, qry->targetList) + { + TargetEntry *tle = (TargetEntry *) lfirst(lc); + Oid atttype = exprType((Node *) tle->expr); + Oid opclass; + + opclass = GetDefaultOpClass(atttype, BTREE_AM_OID); + if (!OidIsValid(opclass)) + ereport(ERROR, + (errcode(ERRCODE_UNDEFINED_OBJECT), + errmsg("data type %s has no default operator class for access method \"%s\"", + format_type_be(atttype), "btree"))); + } + + /* restrictions for rtable */ + foreach(lc, qry->rtable) + { + RangeTblEntry *rte = (RangeTblEntry *) lfirst(lc); + + if (rte->subquery) + ereport(ERROR, + (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg("subquery is not supported on incrementally maintainable materialized view"))); + + if (rte->tablesample != NULL) + ereport(ERROR, + (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg("TABLESAMPLE clause is not supported on incrementally maintainable materialized view"))); + + if (rte->relkind == RELKIND_PARTITIONED_TABLE) + ereport(ERROR, + (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg("partitioned table is not supported on incrementally maintainable materialized view"))); + + if (rte->relkind == RELKIND_RELATION && has_superclass(rte->relid)) + ereport(ERROR, + (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg("partitions is not supported on incrementally maintainable materialized view"))); + + if (rte->relkind == RELKIND_RELATION && find_inheritance_children(rte->relid, NoLock) != NIL) + ereport(ERROR, + (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg("inheritance parent is not supported on incrementally maintainable materialized view"))); + + if (rte->relkind == RELKIND_FOREIGN_TABLE) + ereport(ERROR, + (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg("foreign table is not supported on incrementally maintainable materialized view"))); + + if (rte->relkind == RELKIND_VIEW || + rte->relkind == RELKIND_MATVIEW) + ereport(ERROR, + (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg("VIEW or MATERIALIZED VIEW is not supported on incrementally maintainable materialized view"))); + + if (rte->rtekind == RTE_VALUES) + ereport(ERROR, + (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg("VALUES is not supported on incrementally maintainable materialized view"))); + + } + + query_tree_walker(qry, check_ivm_restriction_walker, NULL, QTW_IGNORE_RANGE_TABLE); + + break; + } + case T_TargetEntry: + { + TargetEntry *tle = (TargetEntry *)node; + if (isIvmName(tle->resname)) + ereport(ERROR, + (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg("column name %s is not supported on incrementally maintainable materialized view", tle->resname))); + + expression_tree_walker(node, check_ivm_restriction_walker, NULL); + break; + } + case T_JoinExpr: + { + JoinExpr *joinexpr = (JoinExpr *)node; + + if (joinexpr->jointype > JOIN_INNER) + ereport(ERROR, + (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg("OUTER JOIN is not supported on incrementally maintainable materialized view"))); + + expression_tree_walker(node, check_ivm_restriction_walker, NULL); + } + break; + case T_Aggref: + ereport(ERROR, + (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg("aggregate function is not supported on incrementally maintainable materialized view"))); + break; + default: + expression_tree_walker(node, check_ivm_restriction_walker, (void *) context); + break; + } + return false; +} + +/* + * CreateIndexOnIMMV + * + * Create a unique index on incremental maintainable materialized view. + * If the view definition query has a GROUP BY clause, the index is created + * on the columns of GROUP BY expressions. Otherwise, if the view contains + * all primary key attributes of its base tables in the target list, the index + * is created on these attributes. In other cases, no index is created. + */ +void +CreateIndexOnIMMV(Query *query, Relation matviewRel) +{ + ListCell *lc; + IndexStmt *index; + ObjectAddress address; + List *constraintList = NIL; + char idxname[NAMEDATALEN]; + List *indexoidlist = RelationGetIndexList(matviewRel); + ListCell *indexoidscan; + Bitmapset *key_attnos; + + snprintf(idxname, sizeof(idxname), "%s_index", RelationGetRelationName(matviewRel)); + + index = makeNode(IndexStmt); + + index->unique = true; + index->primary = false; + index->isconstraint = false; + index->deferrable = false; + index->initdeferred = false; + index->idxname = idxname; + index->relation = + makeRangeVar(get_namespace_name(RelationGetNamespace(matviewRel)), + pstrdup(RelationGetRelationName(matviewRel)), + -1); + index->accessMethod = DEFAULT_INDEX_TYPE; + index->options = NIL; + index->tableSpace = get_tablespace_name(matviewRel->rd_rel->reltablespace); + index->whereClause = NULL; + index->indexParams = NIL; + index->indexIncludingParams = NIL; + index->excludeOpNames = NIL; + index->idxcomment = NULL; + index->indexOid = InvalidOid; + index->oldNumber = InvalidRelFileNumber; + index->oldCreateSubid = InvalidSubTransactionId; + index->oldFirstRelfilelocatorSubid = InvalidSubTransactionId; + index->transformed = true; + index->concurrent = false; + index->if_not_exists = false; + + /* create index on the base tables' primary key columns */ + key_attnos = get_primary_key_attnos_from_query(query, &constraintList); + if (key_attnos) + { + foreach(lc, query->targetList) + { + TargetEntry *tle = (TargetEntry *) lfirst(lc); + Form_pg_attribute attr = TupleDescAttr(matviewRel->rd_att, tle->resno - 1); + + if (bms_is_member(tle->resno - FirstLowInvalidHeapAttributeNumber, key_attnos)) + { + IndexElem *iparam; + + iparam = makeNode(IndexElem); + iparam->name = pstrdup(NameStr(attr->attname)); + iparam->expr = NULL; + iparam->indexcolname = NULL; + iparam->collation = NIL; + iparam->opclass = NIL; + iparam->opclassopts = NIL; + iparam->ordering = SORTBY_DEFAULT; + iparam->nulls_ordering = SORTBY_NULLS_DEFAULT; + index->indexParams = lappend(index->indexParams, iparam); + } + } + } + else + { + /* create no index, just notice that an appropriate index is necessary for efficient IVM */ + ereport(NOTICE, + (errmsg("could not create an index on materialized view \"%s\" automatically", + RelationGetRelationName(matviewRel)), + errdetail("This target list does not have all the primary key columns. "), + errhint("Create an index on the materialized view for efficient incremental maintenance."))); + return; + } + + /* If we have a compatible index, we don't need to create another. */ + foreach(indexoidscan, indexoidlist) + { + Oid indexoid = lfirst_oid(indexoidscan); + Relation indexRel; + bool hasCompatibleIndex = false; + + indexRel = index_open(indexoid, AccessShareLock); + + if (CheckIndexCompatible(indexRel->rd_id, + index->accessMethod, + index->indexParams, + index->excludeOpNames, + false)) + hasCompatibleIndex = true; + + index_close(indexRel, AccessShareLock); + + if (hasCompatibleIndex) + return; + } + + address = DefineIndex(NULL, + RelationGetRelid(matviewRel), + index, + InvalidOid, + InvalidOid, + InvalidOid, + -1, + false, true, false, false, true); + + ereport(NOTICE, + (errmsg("created index \"%s\" on materialized view \"%s\"", + idxname, RelationGetRelationName(matviewRel)))); + + /* + * Make dependencies so that the index is dropped if any base tables' + * primary key is dropped. + */ + foreach(lc, constraintList) + { + Oid constraintOid = lfirst_oid(lc); + ObjectAddress refaddr; + + refaddr.classId = ConstraintRelationId; + refaddr.objectId = constraintOid; + refaddr.objectSubId = 0; + + recordDependencyOn(&address, &refaddr, DEPENDENCY_NORMAL); + } +} + + +/* + * get_primary_key_attnos_from_query + * + * Identify the columns in base tables' primary keys in the target list. + * + * Returns a Bitmapset of the column attnos of the primary key's columns of + * tables that used in the query. The attnos are offset by + * FirstLowInvalidHeapAttributeNumber as same as get_primary_key_attnos. + * + * If any table has no primary key or any primary key's columns is not in + * the target list, return NULL. We also return NULL if any pkey constraint + * is deferrable. + * + * constraintList is set to a list of the OIDs of the pkey constraints. + */ +static Bitmapset * +get_primary_key_attnos_from_query(Query *query, List **constraintList) +{ + List *key_attnos_list = NIL; + ListCell *lc; + int i; + Bitmapset *keys = NULL; + Relids rels_in_from; + + /* + * Collect primary key attributes from all tables used in query. The key attributes + * sets for each table are stored in key_attnos_list in order by RTE index. + */ + foreach(lc, query->rtable) + { + RangeTblEntry *r = (RangeTblEntry*) lfirst(lc); + Bitmapset *key_attnos; + bool has_no_pkey = false; + + /* for tables, call get_primary_key_attnos */ + if (r->rtekind == RTE_RELATION) + { + Oid constraintOid; + key_attnos = get_primary_key_attnos(r->relid, false, &constraintOid); + *constraintList = lappend_oid(*constraintList, constraintOid); + has_no_pkey = (key_attnos == NULL); + } + /* + * Ignore join rels, because they are flatten later by + * flatten_join_alias_vars(). Store NULL into key_attnos_list + * as a dummy. + */ + else if (r->rtekind == RTE_JOIN) + { + key_attnos = NULL; + } + /* for other RTEs, store NULL into key_attnos_list */ + else + has_no_pkey = true; + + /* + * If any table or subquery has no primary key or its pkey constraint is deferrable, + * we cannot get key attributes for this query, so return NULL. + */ + if (has_no_pkey) + return NULL; + + key_attnos_list = lappend(key_attnos_list, key_attnos); + } + + /* Collect key attributes appearing in the target list */ + i = 1; + foreach(lc, query->targetList) + { + TargetEntry *tle = (TargetEntry *) flatten_join_alias_vars(NULL, query, lfirst(lc)); + + if (IsA(tle->expr, Var)) + { + Var *var = (Var*) tle->expr; + Bitmapset *key_attnos = list_nth(key_attnos_list, var->varno - 1); + + /* check if this attribute is from a base table's primary key */ + if (bms_is_member(var->varattno - FirstLowInvalidHeapAttributeNumber, key_attnos)) + { + /* + * Remove found key attributes from key_attnos_list, and add this + * to the result list. + */ + key_attnos = bms_del_member(key_attnos, var->varattno - FirstLowInvalidHeapAttributeNumber); + if (bms_is_empty(key_attnos)) + { + key_attnos_list = list_delete_nth_cell(key_attnos_list, var->varno - 1); + key_attnos_list = list_insert_nth(key_attnos_list, var->varno - 1, NULL); + } + keys = bms_add_member(keys, i - FirstLowInvalidHeapAttributeNumber); + } + } + i++; + } + + /* Collect RTE indexes of relations appearing in the FROM clause */ + rels_in_from = get_relids_in_jointree((Node *) query->jointree, false, false); + + /* + * Check if all key attributes of relations in FROM are appearing in the target + * list. If an attribute remains in key_attnos_list in spite of the table is used + * in FROM clause, the target is missing this key attribute, so we return NULL. + */ + i = 1; + foreach(lc, key_attnos_list) + { + Bitmapset *bms = (Bitmapset *)lfirst(lc); + if (!bms_is_empty(bms) && bms_is_member(i, rels_in_from)) + return NULL; + i++; + } + + return keys; +} diff --git a/src/backend/commands/matview.c b/src/backend/commands/matview.c index f7d8007f796..eaf39f80cd3 100644 --- a/src/backend/commands/matview.c +++ b/src/backend/commands/matview.c @@ -23,24 +23,36 @@ #include "catalog/indexing.h" #include "catalog/namespace.h" #include "catalog/pg_am.h" +#include "catalog/pg_depend.h" +#include "catalog/pg_trigger.h" #include "catalog/pg_opclass.h" #include "commands/matview.h" #include "commands/repack.h" #include "commands/tablecmds.h" #include "commands/tablespace.h" +#include "commands/createas.h" #include "executor/executor.h" #include "executor/spi.h" +#include "executor/tstoreReceiver.h" #include "miscadmin.h" +#include "nodes/makefuncs.h" +#include "parser/analyze.h" +#include "parser/parse_clause.h" +#include "parser/parse_func.h" +#include "parser/parse_relation.h" #include "pgstat.h" #include "rewrite/rewriteHandler.h" +#include "rewrite/rowsecurity.h" #include "storage/lmgr.h" +#include "storage/lwlock.h" #include "tcop/tcopprot.h" #include "utils/builtins.h" +#include "utils/fmgroids.h" #include "utils/lsyscache.h" #include "utils/rel.h" #include "utils/snapmgr.h" #include "utils/syscache.h" - +#include "utils/typcache.h" typedef struct { @@ -53,6 +65,97 @@ typedef struct BulkInsertState bistate; /* bulk insert state */ } DR_transientrel; +#define MV_INIT_QUERYHASHSIZE 16 + +/* + * MV_TriggerHashEntry + * + * Hash entry for base tables on which IVM trigger is invoked + */ +typedef struct MV_TriggerHashEntry +{ + Oid matview_id; /* OID of the materialized view */ + int before_trig_count; /* count of before triggers invoked */ + int after_trig_count; /* count of after triggers invoked */ + + Snapshot snapshot; /* Snapshot just before table change */ + + List *tables; /* List of MV_TriggerTable */ + bool has_old; /* tuples are deleted from any table? */ + bool has_new; /* tuples are inserted into any table? */ + + /* + * List of sub-transaction IDs that incrementally updated the view. + * This list is maintained through a transaction, and an ID is removed + * when a sub-transaction is aborted. If any ID is left when the + * transaction is committed, this means the view is incrementally + * updated in this transaction. + */ + List *subxids; +} MV_TriggerHashEntry; + +/* + * MV_TriggerTable + * + * IVM related data for tables on which the trigger is invoked. + */ +typedef struct MV_TriggerTable +{ + Oid table_id; /* OID of the modified table */ + List *old_tuplestores; /* tuplestores for deleted tuples */ + List *new_tuplestores; /* tuplestores for inserted tuples */ + + List *rte_indexes; /* List of RTE index of the modified table */ + RangeTblEntry *original_rte; /* the original RTE saved before rewriting query */ + + Relation rel; /* relation of the modified table */ + TupleTableSlot *slot; /* for checking visibility in the pre-state table */ +} MV_TriggerTable; + +typedef struct DroppedImmvInfo +{ + Oid immv_oid; + SubTransactionId subxid; +} DroppedImmvInfo; + +static HTAB *mv_trigger_info = NULL; +static HTAB *dropped_immv = NULL; + +static HTAB *LastIvmUpdateHash; + +#define LAST_IVM_UPDATE_HASH_SIZE 256 + +static void IvmShmemRequest(void *arg); + +/* hash table entries */ +typedef struct LastIvmUpdateEntry +{ + Oid oid; /* hash key */ + FullTransactionId last_ivm_update; +} LastIvmUpdateEntry; + +const ShmemCallbacks IvmShmemCallbacks = { + .request_fn = IvmShmemRequest, +}; + +static void +IvmShmemRequest(void *arg) +{ + ShmemRequestHash(.name = "LastIvmUpdate hash", + .nelems = LAST_IVM_UPDATE_HASH_SIZE, + .ptr = &LastIvmUpdateHash, + .hash_info.keysize = sizeof(Oid), + .hash_info.entrysize = sizeof(LastIvmUpdateEntry), + .hash_flags = HASH_ELEM | HASH_BLOBS, + ); +} + +static bool in_delta_calculation = false; + +/* ENR name for materialized view delta */ +#define NEW_DELTA_ENRNAME "new_delta" +#define OLD_DELTA_ENRNAME "old_delta" + static int matview_maintenance_depth = 0; static void transientrel_startup(DestReceiver *self, int operation, TupleDesc typeinfo); @@ -60,6 +163,8 @@ static bool transientrel_receive(TupleTableSlot *slot, DestReceiver *self); static void transientrel_shutdown(DestReceiver *self); static void transientrel_destroy(DestReceiver *self); static uint64 refresh_matview_datafill(DestReceiver *dest, Query *query, + QueryEnvironment *queryEnv, + TupleDesc *resultTupleDesc, const char *queryString, bool is_create); static void refresh_by_match_merge(Oid matviewOid, Oid tempOid, Oid relowner, int save_sec_context); @@ -67,6 +172,41 @@ static void refresh_by_heap_swap(Oid matviewOid, Oid OIDNewHeap, char relpersist static bool is_usable_unique_index(Relation indexRel); static void OpenMatViewIncrementalMaintenance(void); static void CloseMatViewIncrementalMaintenance(void); +static Query *get_matview_query(Relation matviewRel); + +static Query *rewrite_query_for_preupdate_state(Query *query, List *tables, + ParseState *pstate, Oid matviewid); +static void register_delta_ENRs(ParseState *pstate, Query *query, List *tables); +static char *make_subquery_targetlist_from_table(MV_TriggerTable *table); +static char *make_delta_enr_name(const char *prefix, Oid relid, int count); +static RangeTblEntry *get_prestate_rte(RangeTblEntry *rte, MV_TriggerTable *table, + QueryEnvironment *queryEnv, Oid matviewid); +static RangeTblEntry *makeDeltaTable(RangeTblEntry *rte, MV_TriggerTable *table, + bool is_new, QueryEnvironment *queryEnv); +static Query *rewrite_query_for_counting(Query *query, ParseState *pstate); + +static void calc_delta(MV_TriggerTable *table, int rte_index, Query *query, + DestReceiver *dest_old, DestReceiver *dest_new, + TupleDesc *tupdesc_old, TupleDesc *tupdesc_new, + QueryEnvironment *queryEnv); +static Query *rewrite_query_for_postupdate_state(Query *query, MV_TriggerTable *table, int rte_index); + +static void apply_delta(Oid matviewOid, Tuplestorestate *old_tuplestores, Tuplestorestate *new_tuplestores, + TupleDesc tupdesc_old, TupleDesc tupdesc_new, + Query *query); +static void apply_old_delta(const char *matviewname, const char *deltaname_old, + List *keys); +static void apply_new_delta(const char *matviewname, const char *deltaname_new, + StringInfo target_list); +static char *get_matching_condition_string(List *keys); +static void generate_equal(StringInfo querybuf, Oid opttype, + const char *leftop, const char *rightop); + +static void mv_InitHashTables(void); +static void clean_up_IVM_hash_entry(MV_TriggerHashEntry *entry, bool is_abort, + SubTransactionId subxid); +static void setLastUpdateXid(Oid immv_oid, FullTransactionId xid); +static FullTransactionId getLastUpdateXid(Oid immv_oid); /* * SetMatViewPopulatedState @@ -108,6 +248,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 * @@ -166,8 +346,6 @@ RefreshMatViewByOid(Oid matviewOid, bool is_create, bool skipData, QueryCompletion *qc) { Relation matviewRel; - RewriteRule *rule; - List *actions; Query *dataQuery; Oid tableSpace; Oid relowner; @@ -178,6 +356,7 @@ RefreshMatViewByOid(Oid matviewOid, bool is_create, bool skipData, int save_sec_context; int save_nestlevel; ObjectAddress address; + bool oldPopulated; matviewRel = table_open(matviewOid, NoLock); relowner = matviewRel->rd_rel->relowner; @@ -193,6 +372,8 @@ RefreshMatViewByOid(Oid matviewOid, bool is_create, bool skipData, save_nestlevel = NewGUCNestLevel(); RestrictSearchPath(); + oldPopulated = RelationIsPopulated(matviewRel); + /* Make sure it is a materialized view. */ if (matviewRel->rd_rel->relkind != RELKIND_MATVIEW) ereport(ERROR, @@ -213,32 +394,7 @@ RefreshMatViewByOid(Oid matviewOid, bool is_create, bool skipData, errmsg("%s options %s and %s cannot be used together", "REFRESH", "CONCURRENTLY", "WITH NO DATA"))); - /* - * 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 @@ -275,12 +431,6 @@ RefreshMatViewByOid(Oid matviewOid, bool is_create, bool skipData, 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. @@ -310,6 +460,90 @@ RefreshMatViewByOid(Oid matviewOid, bool is_create, bool skipData, relpersistence = matviewRel->rd_rel->relpersistence; } + /* delete IMMV triggers. */ + if (RelationIsIVM(matviewRel) && skipData) + { + Relation tgRel; + Relation depRel; + ScanKeyData key; + SysScanDesc scan; + HeapTuple tup; + ObjectAddresses *immv_triggers; + + immv_triggers = new_object_addresses(); + + tgRel = table_open(TriggerRelationId, RowExclusiveLock); + depRel = table_open(DependRelationId, RowExclusiveLock); + + /* search triggers that depends on IMMV. */ + ScanKeyInit(&key, + Anum_pg_depend_refobjid, + BTEqualStrategyNumber, F_OIDEQ, + ObjectIdGetDatum(matviewOid)); + scan = systable_beginscan(depRel, DependReferenceIndexId, true, + NULL, 1, &key); + while ((tup = systable_getnext(scan)) != NULL) + { + ObjectAddress obj; + Form_pg_depend foundDep = (Form_pg_depend) GETSTRUCT(tup); + + if (foundDep->classid == TriggerRelationId) + { + HeapTuple tgtup; + ScanKeyData tgkey[1]; + SysScanDesc tgscan; + Form_pg_trigger tgform; + + /* Find the trigger name. */ + ScanKeyInit(&tgkey[0], + Anum_pg_trigger_oid, + BTEqualStrategyNumber, F_OIDEQ, + ObjectIdGetDatum(foundDep->objid)); + + tgscan = systable_beginscan(tgRel, TriggerOidIndexId, true, + NULL, 1, tgkey); + tgtup = systable_getnext(tgscan); + if (!HeapTupleIsValid(tgtup)) + elog(ERROR, "could not find tuple for immv trigger %u", foundDep->objid); + + tgform = (Form_pg_trigger) GETSTRUCT(tgtup); + + /* If trigger is created by IMMV, delete it. */ + if (strncmp(NameStr(tgform->tgname), "IVM_trigger_", 12) == 0) + { + obj.classId = foundDep->classid; + obj.objectId = foundDep->objid; + obj.objectSubId = foundDep->refobjsubid; + add_exact_object_address(&obj, immv_triggers); + } + systable_endscan(tgscan); + } + } + systable_endscan(scan); + + performMultipleDeletions(immv_triggers, DROP_RESTRICT, PERFORM_DELETION_INTERNAL); + + table_close(depRel, RowExclusiveLock); + table_close(tgRel, RowExclusiveLock); + free_object_addresses(immv_triggers); + } + + /* + * Create triggers on incremental maintainable materialized view + * This argument should use 'dataQuery'. This needs to use a rewritten query, + * because a sublink in jointree is not supported by this function. + * + * This is performed before generating data because we have to wait + * concurrent transactions modifying a base table and then take a snapshot + * to see changes by these transactions to make sure a consistent view + * is created. + */ + if (RelationIsIVM(matviewRel) && !skipData && !oldPopulated) + { + CreateIvmTriggersOnBaseTables(dataQuery, matviewOid); + CreateIndexOnIMMV(dataQuery, matviewRel); + } + /* * Create the transient table that will receive the regenerated data. Lock * it against access by any other process until commit (by which time it @@ -326,8 +560,43 @@ RefreshMatViewByOid(Oid matviewOid, bool is_create, bool skipData, DestReceiver *dest; dest = CreateTransientRelDestReceiver(OIDNewHeap); - processed = refresh_matview_datafill(dest, dataQuery, queryString, - is_create); + + if (RelationIsIVM(matviewRel)) + { + /* + * In READ COMMITTED, get and push the latest snapshot again to see the + * results of concurrent transactions committed after the current + * transaction started. + */ + if (!IsolationUsesXactSnapshot()) + PushActiveSnapshot(GetTransactionSnapshot()); + + /* + * If a concurrent transaction updated the view incrementally and was + * committed before we acquired the lock, the results of refresh_immv could + * be inconsistent. Therefore, we have to check the transaction ID of the + * most recent update of the view, and if this was in progress at the + * transaction start, raise an error to prevent anomalies. + */ + if (!is_create) + { + FullTransactionId xid; + + xid = getLastUpdateXid(matviewOid); + if (XidInMVCCSnapshot(XidFromFullTransactionId(xid), GetActiveSnapshot())) + ereport(ERROR, + (errcode(ERRCODE_T_R_SERIALIZATION_FAILURE), + errmsg("the materialized view is incrementally updated in concurrent transaction"), + errhint("The transaction might succeed if retried."))); + } + } + + processed = refresh_matview_datafill(dest, dataQuery, NULL, NULL, + queryString, is_create); + + /* Pop the original snapshot. */ + if (RelationIsIVM(matviewRel) && !IsolationUsesXactSnapshot()) + PopActiveSnapshot(); } /* Make the matview match the newly generated data. */ @@ -402,6 +671,8 @@ RefreshMatViewByOid(Oid matviewOid, bool is_create, bool skipData, */ static uint64 refresh_matview_datafill(DestReceiver *dest, Query *query, + QueryEnvironment *queryEnv, + TupleDesc *resultTupleDesc, const char *queryString, bool is_create) { List *rewritten; @@ -439,7 +710,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); @@ -449,6 +720,9 @@ refresh_matview_datafill(DestReceiver *dest, Query *query, processed = queryDesc->estate->es_processed; + if (resultTupleDesc) + *resultTupleDesc = CreateTupleDescCopy(queryDesc->tupDesc); + /* and clean up */ ExecutorFinish(queryDesc); ExecutorEnd(queryDesc); @@ -967,3 +1241,1488 @@ CloseMatViewIncrementalMaintenance(void) matview_maintenance_depth--; Assert(matview_maintenance_depth >= 0); } + +/* + * 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); +} + + +/* ---------------------------------------------------- + * Incremental View Maintenance routines + * --------------------------------------------------- + */ + +/* + * IVM_immediate_before + * + * IVM trigger function invoked before base table is modified. If this is + * invoked firstly in the same statement, we save the transaction id and the + * command id at that time. + */ +Datum +IVM_immediate_before(PG_FUNCTION_ARGS) +{ + TriggerData *trigdata = (TriggerData *) fcinfo->context; + char *matviewOid_text = trigdata->tg_trigger->tgargs[0]; + char *ex_lock_text = trigdata->tg_trigger->tgargs[1]; + Oid matviewOid; + MV_TriggerHashEntry *entry; + bool found; + bool ex_lock; + + matviewOid = DatumGetObjectId(DirectFunctionCall1(oidin, CStringGetDatum(matviewOid_text))); + ex_lock = DatumGetBool(DirectFunctionCall1(boolin, CStringGetDatum(ex_lock_text))); + + /* If the view has more than one tables, we have to use an exclusive lock. */ + if (ex_lock) + { + FullTransactionId xid; + + /* + * Wait for concurrent transactions which update this materialized view at + * READ COMMITED. This is needed to see changes committed in other + * transactions. No wait and raise an error at REPEATABLE READ or + * SERIALIZABLE to prevent update anomalies of matviews. + * XXX: dead-lock is possible here. + */ + if (!IsolationUsesXactSnapshot()) + LockRelationOid(matviewOid, ExclusiveLock); + else if (!ConditionalLockRelationOid(matviewOid, ExclusiveLock)) + { + /* try to throw error by name; relation could be deleted... */ + char *relname = get_rel_name(matviewOid); + + if (!relname) + ereport(ERROR, + (errcode(ERRCODE_LOCK_NOT_AVAILABLE), + errmsg("could not obtain lock on materialized view during incremental maintenance"))); + + ereport(ERROR, + (errcode(ERRCODE_LOCK_NOT_AVAILABLE), + errmsg("could not obtain lock on materialized view \"%s\" during incremental maintenance", + relname))); + } + + /* + * Even if we can acquire an lock, a concurrent transaction could have + * updated the view incrementally and been committed before we acquired + * the lock. Therefore, we have to check the transaction ID of the most + * recent update of the view, and if this was in progress at the + * transaction start, raise an error to prevent anomalies. + */ + xid = getLastUpdateXid(matviewOid); + if (XidInMVCCSnapshot(XidFromFullTransactionId(xid), GetTransactionSnapshot())) + ereport(ERROR, + (errcode(ERRCODE_T_R_SERIALIZATION_FAILURE), + errmsg("the materialized view is incrementally updated in concurrent transaction"), + errhint("The transaction might succeed if retried."))); + } + else + LockRelationOid(matviewOid, RowExclusiveLock); + + /* + * On the first call initialize the hashtable + */ + if (!mv_trigger_info) + mv_InitHashTables(); + + entry = (MV_TriggerHashEntry *) hash_search(mv_trigger_info, + (void *) &matviewOid, + HASH_ENTER, &found); + + /* On the first BEFORE to update the view, initialize trigger data */ + if (!found || entry->snapshot == InvalidSnapshot) + { + Snapshot snapshot; + + /* + * Get a snapshot just before the table was modified for checking + * tuple visibility in the pre-update state of the table. + * + * In READ COMMITTED, use the latest snapshot again to see the + * results of concurrent transactions committed after the current + * transaction started. + */ + if (IsolationUsesXactSnapshot()) + snapshot = GetActiveSnapshot(); + else + snapshot = GetTransactionSnapshot(); + + entry->matview_id = matviewOid; + entry->before_trig_count = 0; + entry->after_trig_count = 0; + entry->snapshot = RegisterSnapshot(snapshot); + entry->tables = NIL; + entry->has_old = false; + entry->has_new = false; + + /* + * If this is the first table modifying query in the transaction, + * initialize the list of subxids. + */ + if (!found) + entry->subxids = NIL; + } + + entry->before_trig_count++; + + return PointerGetDatum(NULL); +} + +/* + * IVM_immediate_maintenance + * + * IVM trigger function invoked after base table is modified. + * For each table, tuplestores of transition tables are collected. + * and after the last modification + */ +Datum +IVM_immediate_maintenance(PG_FUNCTION_ARGS) +{ + TriggerData *trigdata = (TriggerData *) fcinfo->context; + Relation rel; + Oid relid; + Oid matviewOid; + Query *query; + Query *rewritten = NULL; + char *matviewOid_text = trigdata->tg_trigger->tgargs[0]; + Relation matviewRel; + int old_depth = matview_maintenance_depth; + SubTransactionId subxid; + + Oid relowner; + Tuplestorestate *old_tuplestore = NULL; + Tuplestorestate *new_tuplestore = NULL; + DestReceiver *dest_new = NULL, *dest_old = NULL; + Oid save_userid; + int save_sec_context; + int save_nestlevel; + + MV_TriggerHashEntry *entry; + MV_TriggerTable *table; + bool found; + + ParseState *pstate; + QueryEnvironment *queryEnv = create_queryEnv(); + MemoryContext oldcxt; + ListCell *lc; + int i; + + + /* Create a ParseState for rewriting the view definition query */ + pstate = make_parsestate(NULL); + pstate->p_queryEnv = queryEnv; + pstate->p_expr_kind = EXPR_KIND_SELECT_TARGET; + + rel = trigdata->tg_relation; + relid = rel->rd_id; + + matviewOid = DatumGetObjectId(DirectFunctionCall1(oidin, CStringGetDatum(matviewOid_text))); + + /* + * On the first call initialize the hashtable + */ + if (!mv_trigger_info) + mv_InitHashTables(); + + /* get the entry for this materialized view */ + entry = (MV_TriggerHashEntry *) hash_search(mv_trigger_info, + (void *) &matviewOid, + HASH_FIND, &found); + Assert (found && entry != NULL); + entry->after_trig_count++; + + /* search the entry for the modified table and create new entry if not found */ + found = false; + foreach(lc, entry->tables) + { + table = (MV_TriggerTable *) lfirst(lc); + if (table->table_id == relid) + { + found = true; + break; + } + } + if (!found) + { + oldcxt = MemoryContextSwitchTo(TopTransactionContext); + + table = (MV_TriggerTable *) palloc0(sizeof(MV_TriggerTable)); + table->table_id = relid; + table->old_tuplestores = NIL; + table->new_tuplestores = NIL; + table->rte_indexes = NIL; + table->slot = MakeSingleTupleTableSlot(RelationGetDescr(rel), table_slot_callbacks(rel)); + /* We assume we have at least RowExclusiveLock on modified tables. */ + table->rel = table_open(RelationGetRelid(rel), NoLock); + entry->tables = lappend(entry->tables, table); + + MemoryContextSwitchTo(oldcxt); + } + + /* Save the transition tables and make a request to not free immediately */ + if (trigdata->tg_oldtable) + { + oldcxt = MemoryContextSwitchTo(TopTransactionContext); + table->old_tuplestores = lappend(table->old_tuplestores, trigdata->tg_oldtable); + entry->has_old = true; + MemoryContextSwitchTo(oldcxt); + } + if (trigdata->tg_newtable) + { + oldcxt = MemoryContextSwitchTo(TopTransactionContext); + table->new_tuplestores = lappend(table->new_tuplestores, trigdata->tg_newtable); + entry->has_new = true; + MemoryContextSwitchTo(oldcxt); + } + if (entry->has_new || entry->has_old) + { + CmdType cmd; + + if (TRIGGER_FIRED_BY_INSERT(trigdata->tg_event)) + cmd = CMD_INSERT; + else if (TRIGGER_FIRED_BY_DELETE(trigdata->tg_event)) + cmd = CMD_DELETE; + else if (TRIGGER_FIRED_BY_UPDATE(trigdata->tg_event)) + cmd = CMD_UPDATE; + else + elog(ERROR,"unsupported trigger type"); + + /* Prolong lifespan of transition tables to the end of the last AFTER trigger */ + SetTransitionTablePreserved(relid, cmd); + } + + + /* If this is not the last AFTER trigger call, immediately exit. */ + Assert (entry->before_trig_count >= entry->after_trig_count); + if (entry->before_trig_count != entry->after_trig_count) + return PointerGetDatum(NULL); + + /* + * If this is the last AFTER trigger call, continue and update the view. + */ + + /* record the subxid that updated the view incrementally */ + subxid = GetCurrentSubTransactionId(); + if (!list_member_xid(entry->subxids, subxid)) + { + oldcxt = MemoryContextSwitchTo(TopTransactionContext); + entry->subxids = lappend_xid(entry->subxids, subxid); + MemoryContextSwitchTo(oldcxt); + } + + /* + * Advance command counter to make the updated base table rows locally + * visible. + */ + CommandCounterIncrement(); + + matviewRel = table_open(matviewOid, NoLock); + + /* Make sure it is a materialized view. */ + Assert(matviewRel->rd_rel->relkind == RELKIND_MATVIEW); + + /* + * In READ COMMITTED, get and push the latest snapshot again to see the + * results of concurrent transactions committed after the current + * transaction started. + */ + if (!IsolationUsesXactSnapshot()) + PushActiveSnapshot(GetTransactionSnapshot()); + + /* + * Check for active uses of the relation in the current transaction, such + * as open scans. + * + * NB: We count on this to protect us against problems with refreshing the + * data using TABLE_INSERT_FROZEN. + */ + CheckTableNotInUse(matviewRel, "refresh a materialized view incrementally"); + + /* + * 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. + * We will switch modes when we are about to execute user code. + */ + relowner = matviewRel->rd_rel->relowner; + GetUserIdAndSecContext(&save_userid, &save_sec_context); + SetUserIdAndSecContext(relowner, + save_sec_context | SECURITY_RESTRICTED_OPERATION); + save_nestlevel = NewGUCNestLevel(); + + /* get view query*/ + query = get_matview_query(matviewRel); + + /* + * When a base table is truncated, the view content will be empty if the + * view definition query does not contain an aggregate without a GROUP clause. + * Therefore, such views can be truncated. + */ + if (TRIGGER_FIRED_BY_TRUNCATE(trigdata->tg_event)) + { + ExecuteTruncateGuts(list_make1(matviewRel), list_make1_oid(matviewOid), + NIL, DROP_RESTRICT, false, false); + + /* Clean up hash entry and delete tuplestores */ + clean_up_IVM_hash_entry(entry, false, InvalidSubTransactionId); + + /* Pop the original snapshot. */ + if (!IsolationUsesXactSnapshot()) + 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); + } + + /* + * rewrite query for calculating deltas + */ + + rewritten = copyObject(query); + + /* Replace resnames in a target list with materialized view's attnames */ + i = 0; + foreach (lc, rewritten->targetList) + { + TargetEntry *tle = (TargetEntry *) lfirst(lc); + Form_pg_attribute attr = TupleDescAttr(matviewRel->rd_att, i); + char *resname = NameStr(attr->attname); + + tle->resname = pstrdup(resname); + i++; + } + + /* Set all tables in the query to pre-update state */ + rewritten = rewrite_query_for_preupdate_state(rewritten, entry->tables, + pstate, matviewOid); + /* Rewrite for counting duplicated tuples */ + rewritten = rewrite_query_for_counting(rewritten, pstate); + + /* Create tuplestores to store view deltas */ + if (entry->has_old) + { + oldcxt = MemoryContextSwitchTo(TopTransactionContext); + + old_tuplestore = tuplestore_begin_heap(false, false, work_mem); + dest_old = CreateDestReceiver(DestTuplestore); + SetTuplestoreDestReceiverParams(dest_old, + old_tuplestore, + TopTransactionContext, + false, + NULL, + NULL); + + MemoryContextSwitchTo(oldcxt); + } + if (entry->has_new) + { + oldcxt = MemoryContextSwitchTo(TopTransactionContext); + + new_tuplestore = tuplestore_begin_heap(false, false, work_mem); + dest_new = CreateDestReceiver(DestTuplestore); + SetTuplestoreDestReceiverParams(dest_new, + new_tuplestore, + TopTransactionContext, + false, + NULL, + NULL); + MemoryContextSwitchTo(oldcxt); + } + + /* for all modified tables */ + foreach(lc, entry->tables) + { + ListCell *lc2; + + table = (MV_TriggerTable *) lfirst(lc); + + /* loop for self-join */ + foreach(lc2, table->rte_indexes) + { + int rte_index = lfirst_int(lc2); + TupleDesc tupdesc_old; + TupleDesc tupdesc_new; + + /* calculate delta tables */ + calc_delta(table, rte_index, rewritten, dest_old, dest_new, + &tupdesc_old, &tupdesc_new, queryEnv); + + /* Set the table in the query to post-update state */ + rewritten = rewrite_query_for_postupdate_state(rewritten, table, rte_index); + + PG_TRY(); + { + /* apply the delta tables to the materialized view */ + apply_delta(matviewOid, old_tuplestore, new_tuplestore, + tupdesc_old, tupdesc_new, query); + } + PG_CATCH(); + { + matview_maintenance_depth = old_depth; + PG_RE_THROW(); + } + PG_END_TRY(); + + /* clear view delta tuplestores */ + if (old_tuplestore) + tuplestore_clear(old_tuplestore); + if (new_tuplestore) + tuplestore_clear(new_tuplestore); + } + } + + /* Clean up hash entry and delete tuplestores */ + clean_up_IVM_hash_entry(entry, false, InvalidSubTransactionId); + if (old_tuplestore) + { + dest_old->rDestroy(dest_old); + tuplestore_end(old_tuplestore); + } + if (new_tuplestore) + { + dest_new->rDestroy(dest_new); + tuplestore_end(new_tuplestore); + } + + /* Pop the original snapshot. */ + if (!IsolationUsesXactSnapshot()) + 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); +} + +/* + * rewrite_query_for_preupdate_state + * + * Rewrite the query so that base tables' RTEs will represent "pre-update" + * state of tables. This is necessary to calculate view delta after multiple + * tables are modified. + */ +static Query* +rewrite_query_for_preupdate_state(Query *query, List *tables, + ParseState *pstate, Oid matviewid) +{ + ListCell *lc; + int num_rte = list_length(query->rtable); + int i; + + + /* register delta ENRs */ + register_delta_ENRs(pstate, query, tables); + + /* XXX: Is necessary? Is this right timing? */ + AcquireRewriteLocks(query, true, false); + + i = 1; + foreach(lc, query->rtable) + { + RangeTblEntry *r = (RangeTblEntry*) lfirst(lc); + + ListCell *lc2; + foreach(lc2, tables) + { + MV_TriggerTable *table = (MV_TriggerTable *) lfirst(lc2); + /* + * if the modified table is found then replace the original RTE with + * "pre-state" RTE and append its index to the list. + */ + if (r->relid == table->table_id) + { + List *securityQuals; + List *withCheckOptions; + bool hasRowSecurity; + bool hasSubLinks; + + RangeTblEntry *rte_pre = get_prestate_rte(r, table, pstate->p_queryEnv, matviewid); + + /* + * Set a row security poslicies of the modified table to the subquery RTE which + * represents the pre-update state of the table. + */ + get_row_security_policies(query, table->original_rte, i, + &securityQuals, &withCheckOptions, + &hasRowSecurity, &hasSubLinks); + + if (hasRowSecurity) + { + query->hasRowSecurity = true; + rte_pre->security_barrier = true; + } + if (hasSubLinks) + query->hasSubLinks = true; + + rte_pre->securityQuals = securityQuals; + lfirst(lc) = rte_pre; + + table->rte_indexes = lappend_int(table->rte_indexes, i); + break; + } + } + + /* finish the loop if we processed all RTE included in the original query */ + if (i++ >= num_rte) + break; + } + + return query; +} + +/* + * register_delta_ENRs + * + * For all modified tables, make ENRs for their transition tables + * and register them to the queryEnv. ENR's RTEs are also appended + * into the list in query tree. + */ +static void +register_delta_ENRs(ParseState *pstate, Query *query, List *tables) +{ + QueryEnvironment *queryEnv = pstate->p_queryEnv; + ListCell *lc; + RangeTblEntry *rte; + + foreach(lc, tables) + { + MV_TriggerTable *table = (MV_TriggerTable *) lfirst(lc); + ListCell *lc2; + int count; + + count = 0; + foreach(lc2, table->old_tuplestores) + { + Tuplestorestate *oldtable = (Tuplestorestate *) lfirst(lc2); + EphemeralNamedRelation enr = + palloc(sizeof(EphemeralNamedRelationData)); + ParseNamespaceItem *nsitem; + + enr->md.name = make_delta_enr_name("old", table->table_id, count); + enr->md.reliddesc = table->table_id; + enr->md.tupdesc = NULL; + enr->md.enrtype = ENR_NAMED_TUPLESTORE; + enr->md.enrtuples = tuplestore_tuple_count(oldtable); + enr->reldata = oldtable; + register_ENR(queryEnv, enr); + + nsitem = addRangeTableEntryForENR(pstate, makeRangeVar(NULL, enr->md.name, -1), true); + rte = nsitem->p_rte; + + query->rtable = lappend(query->rtable, rte); + + count++; + } + + count = 0; + foreach(lc2, table->new_tuplestores) + { + Tuplestorestate *newtable = (Tuplestorestate *) lfirst(lc2); + EphemeralNamedRelation enr = + palloc(sizeof(EphemeralNamedRelationData)); + ParseNamespaceItem *nsitem; + + enr->md.name = make_delta_enr_name("new", table->table_id, count); + enr->md.reliddesc = table->table_id; + enr->md.tupdesc = NULL; + enr->md.enrtype = ENR_NAMED_TUPLESTORE; + enr->md.enrtuples = tuplestore_tuple_count(newtable); + enr->reldata = newtable; + register_ENR(queryEnv, enr); + + nsitem = addRangeTableEntryForENR(pstate, makeRangeVar(NULL, enr->md.name, -1), true); + rte = nsitem->p_rte; + + query->rtable = lappend(query->rtable, rte); + + count++; + } + } +} + +#define DatumGetItemPointer(X) ((ItemPointer) DatumGetPointer(X)) +#define PG_GETARG_ITEMPOINTER(n) DatumGetItemPointer(PG_GETARG_DATUM(n)) + +/* + * ivm_visible_in_prestate + * + * Check visibility of a tuple specified by the tableoid and item pointer + * using the snapshot taken just before the table was modified. + */ +Datum +ivm_visible_in_prestate(PG_FUNCTION_ARGS) +{ + Oid tableoid = PG_GETARG_OID(0); + ItemPointer itemPtr = PG_GETARG_ITEMPOINTER(1); + Oid matviewOid = PG_GETARG_OID(2); + MV_TriggerHashEntry *entry; + MV_TriggerTable *table = NULL; + ListCell *lc; + bool found; + bool result; + + if (!in_delta_calculation) + ereport(ERROR, + (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg("ivm_visible_in_prestate can be called only in delta calculation"))); + + entry = (MV_TriggerHashEntry *) hash_search(mv_trigger_info, + (void *) &matviewOid, + HASH_FIND, &found); + Assert (found && entry != NULL); + + foreach(lc, entry->tables) + { + table = (MV_TriggerTable *) lfirst(lc); + if (table->table_id == tableoid) + break; + } + + Assert (table != NULL); + + result = table_tuple_fetch_row_version(table->rel, itemPtr, entry->snapshot, table->slot); + + PG_RETURN_BOOL(result); +} + +/* + * get_prestate_rte + * + * Rewrite RTE of the modified table to a subquery which represents + * "pre-state" table. The original RTE is saved in table->rte_original. + */ +static RangeTblEntry* +get_prestate_rte(RangeTblEntry *rte, MV_TriggerTable *table, + QueryEnvironment *queryEnv, Oid matviewid) +{ + StringInfoData str; + RawStmt *raw; + Query *subquery; + ParseState *pstate; + char *relname; + static char *subquery_tl; + int i; + + pstate = make_parsestate(NULL); + pstate->p_queryEnv = queryEnv; + pstate->p_expr_kind = EXPR_KIND_SELECT_TARGET; + + relname = quote_qualified_identifier( + get_namespace_name(RelationGetNamespace(table->rel)), + RelationGetRelationName(table->rel)); + + subquery_tl = make_subquery_targetlist_from_table(table); + + /* + * Filtering inserted row using the snapshot taken before the table + * is modified. ctid is required for maintaining outer join views. + */ + initStringInfo(&str); + appendStringInfo(&str, + "SELECT %s FROM %s t" + " WHERE pg_catalog.ivm_visible_in_prestate(t.tableoid, t.ctid ,%d::pg_catalog.oid)", + subquery_tl, relname, matviewid); + + /* + * Re-add rows deleted from the old transition tables, excluding those + * also present in the new transition tables. + */ + if (list_length(table->old_tuplestores) > 0) + { + appendStringInfo(&str," UNION ALL SELECT %s FROM (", + subquery_tl); + + for (i = 0; i < list_length(table->old_tuplestores); i++) + { + if (i != 0) + appendStringInfo(&str, " UNION ALL "); + appendStringInfo(&str," TABLE %s", + make_delta_enr_name("old", table->table_id, i)); + } + for (i = 0; i < list_length(table->new_tuplestores); i++) + { + appendStringInfo(&str, " EXCEPT ALL "); + appendStringInfo(&str," TABLE %s", + make_delta_enr_name("new", table->table_id, i)); + } + appendStringInfo(&str,")"); + } + + /* Get a subquery representing pre-state of the table */ + raw = (RawStmt*)linitial(raw_parser(str.data, RAW_PARSE_DEFAULT)); + subquery = transformStmt(pstate, raw->stmt); + + /* save the original RTE */ + table->original_rte = copyObject(rte); + + rte->rtekind = RTE_SUBQUERY; + rte->subquery = subquery; + rte->security_barrier = false; + + /* Clear fields that should not be set in a subquery RTE */ + rte->relid = InvalidOid; + rte->relkind = 0; + rte->rellockmode = 0; + rte->tablesample = NULL; + rte->perminfoindex = 0; /* no permission checking for this RTE */ + rte->inh = false; /* must not be set for a subquery */ + + return rte; +} + +/* + * make_subquery_targetlist_from_table + * + * Make a targetlist string of a subquery representing a delta table or a + * pre-update state table. This subquery substitutes a modified table RTE + * in the view definition query during view maintenance. In the targetlist, + * column names appear in order of the table definition. However, for + * attribute numbers of vars in the query tree to reference columns of the + * subquery correctly even though the table has a dropped column, put "null" + * as a dummy value at the position of a dropped column. + * + * We would also able to walk the query tree to rewrite varattnos, but + * crafting targetlist is more simple and reasonable. + */ +static char* +make_subquery_targetlist_from_table(MV_TriggerTable *table) +{ + StringInfoData str; + TupleDesc tupdesc; + int i; + + tupdesc = RelationGetDescr(table->rel); + initStringInfo(&str); + for (i = 0; i < tupdesc->natts; i++) + { + Form_pg_attribute attr = TupleDescAttr(tupdesc, i); + + if (i > 0) + appendStringInfo(&str, ", "); + + if (attr->attisdropped) + appendStringInfo(&str, "null"); + else + appendStringInfo(&str, "%s", quote_identifier(NameStr(attr->attname))); + } + + return str.data; +} + +/* + * make_delta_enr_name + * + * Make a name for ENR of a transition table from the base table's oid. + * prefix will be "new" or "old" depending on its transition table kind.. + */ +static char* +make_delta_enr_name(const char *prefix, Oid relid, int count) +{ + char buf[NAMEDATALEN]; + char *name; + + snprintf(buf, NAMEDATALEN, "__ivm_%s_%u_%u", prefix, relid, count); + name = pstrdup(buf); + + return name; +} + +/* + * makeDeltaTable + * + * Make a RTE representing a delta of the specified table. + */ +static RangeTblEntry* +makeDeltaTable(RangeTblEntry *rte, MV_TriggerTable *table, + bool is_new, QueryEnvironment *queryEnv) +{ + StringInfoData str; + ParseState *pstate; + RawStmt *raw; + Query *sub; + int i; + + const char *prefix_union = is_new ? "new" : "old"; + const char *prefix_except = is_new ? "old" : "new"; + int num_union = is_new ? list_length(table->new_tuplestores) : list_length(table->old_tuplestores); + int num_except = is_new ? list_length(table->old_tuplestores) : list_length(table->new_tuplestores); + + /* the previous RTE must be a subquery which represents "pre-state" table */ + Assert(rte->rtekind == RTE_SUBQUERY); + + /* Create a ParseState for rewriting the view definition query */ + pstate = make_parsestate(NULL); + pstate->p_queryEnv = queryEnv; + pstate->p_expr_kind = EXPR_KIND_NONE; + + initStringInfo(&str); + appendStringInfo(&str, + " SELECT %s FROM ( ", + make_subquery_targetlist_from_table(table)); + + for (i = 0; i < num_union; i++) + { + if (i > 0) + appendStringInfo(&str, " UNION ALL "); + + appendStringInfo(&str, + " TABLE %s", + make_delta_enr_name(prefix_union, table->table_id, i)); + } + for (i = 0; i < num_except; i++) + { + appendStringInfo(&str, " EXCEPT ALL "); + + appendStringInfo(&str, + " TABLE %s", + make_delta_enr_name(prefix_except, table->table_id, i)); + } + appendStringInfo(&str,")"); + + raw = (RawStmt*)linitial(raw_parser(str.data, RAW_PARSE_DEFAULT)); + sub = transformStmt(pstate, raw->stmt); + + /* + * Update the subquery so that it represent the combined transition + * table. Note that we leave the security_barrier and securityQuals + * fields so that the subquery relation can be protected by the RLS + * policy as same as the modified table. + */ + rte->rtekind = RTE_SUBQUERY; + rte->subquery = sub; + + return rte; +} + +/* + * rewrite_query_for_counting + * + * Rewrite query for counting duplicated tuples. + */ +static Query * +rewrite_query_for_counting(Query *query, ParseState *pstate) +{ + TargetEntry *tle_count; + FuncCall *fn; + Node *node; + + /* Add count(*) for counting distinct tuples in views */ + fn = makeFuncCall(SystemFuncName("count"), NIL, COERCE_EXPLICIT_CALL, -1); + fn->agg_star = true; + if (!query->groupClause && !query->hasAggs) + query->groupClause = transformDistinctClause(NULL, &query->targetList, query->sortClause, false); + + node = ParseFuncOrColumn(pstate, fn->funcname, NIL, NULL, fn, false, -1); + + tle_count = makeTargetEntry((Expr *) node, + list_length(query->targetList) + 1, + pstrdup("__ivm_count__"), + false); + query->targetList = lappend(query->targetList, tle_count); + query->hasAggs = true; + + return query; +} + +/* + * calc_delta + * + * Calculate view deltas generated under the modification of a table specified + * by the RTE index. + */ +static void +calc_delta(MV_TriggerTable *table, int rte_index, Query *query, + DestReceiver *dest_old, DestReceiver *dest_new, + TupleDesc *tupdesc_old, TupleDesc *tupdesc_new, + QueryEnvironment *queryEnv) +{ + ListCell *lc = list_nth_cell(query->rtable, rte_index - 1); + RangeTblEntry *rte = (RangeTblEntry *) lfirst(lc); + + in_delta_calculation = true; + + /* Generate old delta */ + if (dest_old && list_length(table->old_tuplestores) > 0) + { + /* Replace the modified table with the old delta table and calculate the old view delta. */ + lfirst(lc) = makeDeltaTable(rte, table, false, queryEnv); + refresh_matview_datafill(dest_old, query, queryEnv, tupdesc_old, "", false); + } + + /* Generate new delta */ + if (dest_new && list_length(table->new_tuplestores) > 0) + { + /* Replace the modified table with the new delta table and calculate the new view delta*/ + lfirst(lc) = makeDeltaTable(rte, table, true, queryEnv); + refresh_matview_datafill(dest_new, query, queryEnv, tupdesc_new, "", false); + } + + in_delta_calculation = false; +} + +/* + * rewrite_query_for_postupdate_state + * + * Rewrite the query so that the specified base table's RTEs will represent + * "post-update" state of tables. This is called after the view delta + * calculation due to changes on this table finishes. + */ +static Query* +rewrite_query_for_postupdate_state(Query *query, MV_TriggerTable *table, int rte_index) +{ + ListCell *lc = list_nth_cell(query->rtable, rte_index - 1); + + /* Retore the original RTE */ + lfirst(lc) = table->original_rte; + + return query; +} + +/* + * apply_delta + * + * Apply deltas to the materialized view. In outer join cases, this requires + * the view maintenance graph. + */ +static void +apply_delta(Oid matviewOid, Tuplestorestate *old_tuplestores, Tuplestorestate *new_tuplestores, + TupleDesc tupdesc_old, TupleDesc tupdesc_new, + Query *query) +{ + StringInfoData querybuf; + StringInfoData target_list_buf; + Relation matviewRel; + char *matviewname; + ListCell *lc; + int i; + List *keys = NIL; + + + /* + * get names of the materialized view and delta tables + */ + + matviewRel = table_open(matviewOid, NoLock); + matviewname = quote_qualified_identifier(get_namespace_name(RelationGetNamespace(matviewRel)), + RelationGetRelationName(matviewRel)); + + /* + * Build parts of the maintenance queries + */ + + initStringInfo(&querybuf); + initStringInfo(&target_list_buf); + + /* build string of target list */ + for (i = 0; i < matviewRel->rd_att->natts; i++) + { + Form_pg_attribute attr = TupleDescAttr(matviewRel->rd_att, i); + char *resname = NameStr(attr->attname); + + if (i != 0) + appendStringInfo(&target_list_buf, ", "); + appendStringInfo(&target_list_buf, "%s", quote_qualified_identifier(NULL, resname)); + } + + i = 0; + foreach (lc, query->targetList) + { + TargetEntry *tle = (TargetEntry *) lfirst(lc); + Form_pg_attribute attr = TupleDescAttr(matviewRel->rd_att, i); + + i++; + + if (tle->resjunk) + continue; + + keys = lappend(keys, attr); + } + + /* Start maintaining the materialized view. */ + OpenMatViewIncrementalMaintenance(); + + /* Open SPI context. */ + if (SPI_connect() != SPI_OK_CONNECT) + elog(ERROR, "SPI_connect failed"); + + /* For tuple deletion */ + if (old_tuplestores && tuplestore_tuple_count(old_tuplestores) > 0) + { + EphemeralNamedRelation enr = palloc(sizeof(EphemeralNamedRelationData)); + int rc; + + /* convert tuplestores to ENR, and register for SPI */ + enr->md.name = pstrdup(OLD_DELTA_ENRNAME); + enr->md.reliddesc = InvalidOid; + enr->md.tupdesc = tupdesc_old; + enr->md.enrtype = ENR_NAMED_TUPLESTORE; + enr->md.enrtuples = tuplestore_tuple_count(old_tuplestores); + enr->reldata = old_tuplestores; + + rc = SPI_register_relation(enr); + if (rc != SPI_OK_REL_REGISTER) + elog(ERROR, "SPI_register failed"); + + apply_old_delta(matviewname, OLD_DELTA_ENRNAME, keys); + + } + /* For tuple insertion */ + if (new_tuplestores && tuplestore_tuple_count(new_tuplestores) > 0) + { + EphemeralNamedRelation enr = palloc(sizeof(EphemeralNamedRelationData)); + int rc; + + /* convert tuplestores to ENR, and register for SPI */ + enr->md.name = pstrdup(NEW_DELTA_ENRNAME); + enr->md.reliddesc = InvalidOid; + enr->md.tupdesc = tupdesc_new;; + enr->md.enrtype = ENR_NAMED_TUPLESTORE; + enr->md.enrtuples = tuplestore_tuple_count(new_tuplestores); + enr->reldata = new_tuplestores; + + rc = SPI_register_relation(enr); + if (rc != SPI_OK_REL_REGISTER) + elog(ERROR, "SPI_register failed"); + + /* apply new delta */ + apply_new_delta(matviewname, NEW_DELTA_ENRNAME, &target_list_buf); + } + + /* We're done maintaining the materialized view. */ + CloseMatViewIncrementalMaintenance(); + + table_close(matviewRel, NoLock); + + /* Close SPI context. */ + if (SPI_finish() != SPI_OK_FINISH) + elog(ERROR, "SPI_finish failed"); +} + +/* + * apply_old_delta + * + * Execute a query for applying a delta table given by deltname_old + * which contains tuples to be deleted from to a materialized view given by + * matviewname. This is used when counting is not required. + */ +static void +apply_old_delta(const char *matviewname, const char *deltaname_old, + List *keys) +{ + StringInfoData querybuf; + StringInfoData keysbuf; + char *match_cond; + ListCell *lc; + + /* build WHERE condition for searching tuples to be deleted */ + match_cond = get_matching_condition_string(keys); + + /* build string of keys list */ + initStringInfo(&keysbuf); + foreach (lc, keys) + { + Form_pg_attribute attr = (Form_pg_attribute) lfirst(lc); + char *resname = NameStr(attr->attname); + appendStringInfo(&keysbuf, "%s", quote_qualified_identifier("mv", resname)); + if (lnext(keys, lc)) + appendStringInfo(&keysbuf, ", "); + } + + /* Search for matching tuples from the view and update or delete if found. */ + initStringInfo(&querybuf); + appendStringInfo(&querybuf, + "DELETE FROM %s WHERE ctid IN (" + "SELECT tid FROM (SELECT pg_catalog.row_number() over (partition by %s) AS \"__ivm_row_number__\"," + "mv.ctid AS tid," + "diff.\"__ivm_count__\"" + "FROM %s AS mv, %s AS diff " + "WHERE %s) v " + "WHERE v.\"__ivm_row_number__\" OPERATOR(pg_catalog.<=) v.\"__ivm_count__\")", + matviewname, + keysbuf.data, + matviewname, deltaname_old, + match_cond); + + if (SPI_exec(querybuf.data, 0) != SPI_OK_DELETE) + elog(ERROR, "SPI_exec failed: %s", querybuf.data); +} + +/* + * apply_new_delta + * + * Execute a query for applying a delta table given by deltname_new + * which contains tuples to be inserted into a materialized view given by + * matviewname. This is used when counting is not required. + */ +static void +apply_new_delta(const char *matviewname, const char *deltaname_new, + StringInfo target_list) +{ + StringInfoData querybuf; + + /* Search for matching tuples from the view and update or delete if found. */ + initStringInfo(&querybuf); + appendStringInfo(&querybuf, + "INSERT INTO %s (%s) SELECT %s FROM (" + "SELECT diff.*, pg_catalog.generate_series(1, diff.\"__ivm_count__\")" + " AS __ivm_generate_series__ " + "FROM %s AS diff) AS v", + matviewname, target_list->data, target_list->data, + deltaname_new); + + if (SPI_exec(querybuf.data, 0) != SPI_OK_INSERT) + elog(ERROR, "SPI_exec failed: %s", querybuf.data); +} + +/* + * get_matching_condition_string + * + * Build a predicate string for looking for a tuple with given keys. + */ +static char * +get_matching_condition_string(List *keys) +{ + StringInfoData match_cond; + ListCell *lc; + + /* If there is no key columns, the condition is always true. */ + if (keys == NIL) + return "true"; + + initStringInfo(&match_cond); + foreach (lc, keys) + { + Form_pg_attribute attr = (Form_pg_attribute) lfirst(lc); + char *resname = NameStr(attr->attname); + char *mv_resname = quote_qualified_identifier("mv", resname); + char *diff_resname = quote_qualified_identifier("diff", resname); + Oid typid = attr->atttypid; + + /* Considering NULL values, we can not use simple = operator. */ + appendStringInfo(&match_cond, "("); + generate_equal(&match_cond, typid, mv_resname, diff_resname); + appendStringInfo(&match_cond, " OR (%s IS NULL AND %s IS NULL))", + mv_resname, diff_resname); + + if (lnext(keys, lc)) + appendStringInfo(&match_cond, " AND "); + } + + return match_cond.data; +} + +/* + * generate_equals + * + * Generate an equality clause using given operands' default equality + * operator. + */ +static void +generate_equal(StringInfo querybuf, Oid opttype, + const char *leftop, const char *rightop) +{ + TypeCacheEntry *typentry; + + typentry = lookup_type_cache(opttype, TYPECACHE_EQ_OPR); + if (!OidIsValid(typentry->eq_opr)) + ereport(ERROR, + (errcode(ERRCODE_UNDEFINED_FUNCTION), + errmsg("could not identify an equality operator for type %s", + format_type_be_qualified(opttype)))); + + generate_operator_clause(querybuf, + leftop, opttype, + typentry->eq_opr, + rightop, opttype); +} + +/* + * mv_InitHashTables + */ +static void +mv_InitHashTables(void) +{ + HASHCTL ctl; + + memset(&ctl, 0, sizeof(ctl)); + ctl.keysize = sizeof(Oid); + ctl.entrysize = sizeof(MV_TriggerHashEntry); + mv_trigger_info = hash_create("MV trigger info", + MV_INIT_QUERYHASHSIZE, + &ctl, HASH_ELEM | HASH_BLOBS); + + memset(&ctl, 0, sizeof(ctl)); + ctl.keysize = sizeof(Oid); + ctl.entrysize = sizeof(DroppedImmvInfo); + dropped_immv = hash_create("Dropped IMMVs", 16, + &ctl, HASH_ELEM | HASH_BLOBS); +} + +/* + * AtAbort_IVM + * + * Clean up hash entries for all materialized views. This is called at + * (sub-)transaction abort. When the top-level transaction is aborted, + * InvalidSubTransactionId is set to subxid. + * + * Also, remove dropped IMMV information if it is aborted. + */ +void +AtAbort_IVM(SubTransactionId subxid) +{ + HASH_SEQ_STATUS seq; + + if (mv_trigger_info) + { + MV_TriggerHashEntry *entry; + hash_seq_init(&seq, mv_trigger_info); + while ((entry = hash_seq_search(&seq)) != NULL) + clean_up_IVM_hash_entry(entry, true, subxid); + } + + if (dropped_immv) + { + DroppedImmvInfo *entry; + hash_seq_init(&seq, dropped_immv); + while ((entry = hash_seq_search(&seq)) != NULL) + { + if (subxid == InvalidSubTransactionId || subxid == entry->subxid) + hash_search(dropped_immv, (void *) &entry->immv_oid, HASH_REMOVE, NULL); + } + } + + in_delta_calculation = false; +} + +/* + * AtPreCommit_IVM + * + * Record the transaction ID that updated the view incrementally. + * Also, remove the entry for dropped IMMVs. + */ +void +AtPreCommit_IVM(void) +{ + HASH_SEQ_STATUS seq; + + if (mv_trigger_info) + { + MV_TriggerHashEntry *entry; + + /* + * Record the transaction ID that udpate the view incrementally, + * and perform the final clean up of the entry. + */ + hash_seq_init(&seq, mv_trigger_info); + while ((entry = hash_seq_search(&seq)) != NULL) + { + setLastUpdateXid(entry->matview_id, GetTopFullTransactionId()); + hash_search(mv_trigger_info, (void *) &entry->matview_id, HASH_REMOVE, NULL); + } + } + + if (dropped_immv) + { + DroppedImmvInfo *entry; + + LWLockAcquire(IvmControlLock, LW_EXCLUSIVE); + hash_seq_init(&seq, dropped_immv); + while ((entry = hash_seq_search(&seq)) != NULL) + { + hash_search(LastIvmUpdateHash, (void *) &entry->immv_oid, HASH_REMOVE, NULL); + hash_search(dropped_immv, (void *) &entry->immv_oid, HASH_REMOVE, NULL); + } + LWLockRelease(IvmControlLock); + } + + in_delta_calculation = false; +} + +/* + * clean_up_IVM_hash_entry + * + * Clean up tuple stores and hash entries for a materialized view after its + * maintenance finished. This is called at the end of table modifying query + * or (sub-)transaction abort. When the top-level transaction is aborted, + * InvalidSubTransactionId is set to subxid. + */ +static void +clean_up_IVM_hash_entry(MV_TriggerHashEntry *entry, bool is_abort, + SubTransactionId subxid) +{ + bool found; + ListCell *lc; + + /* clean up tuple stores */ + foreach(lc, entry->tables) + { + MV_TriggerTable *table = (MV_TriggerTable *) lfirst(lc); + + list_free(table->old_tuplestores); + list_free(table->new_tuplestores); + if (!is_abort) + { + ExecDropSingleTupleTableSlot(table->slot); + table_close(table->rel, NoLock); + } + } + list_free(entry->tables); + entry->tables = NIL; + + if (is_abort) + { + bool remove_entry = false; + + /* + * When the top-level transaction is aborted, remove all subxids. + * When a sub-transaction is aborted, remove only its subxid. + */ + if (subxid == InvalidSubTransactionId) + remove_entry = true; + else + { + foreach(lc, entry->subxids) + { + if (lfirst_xid(lc) == subxid) + { + entry->subxids = list_delete_cell(entry->subxids, lc); + break; + } + } + + /* + * If all the subxid are removed, it means that the view was not + * updated at all in this transaction. + */ + if (list_length(entry->subxids) == 0) + remove_entry = true; + } + + /* + * Remove entries of not updated views from the hash table. + */ + if (remove_entry) + hash_search(mv_trigger_info, (void *) &entry->matview_id, HASH_REMOVE, &found); + } + else + { + /* When the query sucsessully finished, unregister the snapshot */ + UnregisterSnapshot(entry->snapshot); + } + + entry->snapshot = InvalidSnapshot; +} + +/* + * setLastUpdateXid + * + * Store the transaction ID that updated the view incremenally. + */ +static void +setLastUpdateXid(Oid immv_oid, FullTransactionId xid) +{ + LastIvmUpdateEntry *entry; + + LWLockAcquire(IvmControlLock, LW_EXCLUSIVE); + entry = (LastIvmUpdateEntry *) hash_search(LastIvmUpdateHash, (void *) &immv_oid, HASH_ENTER, NULL); + entry->oid = immv_oid; + entry->last_ivm_update = xid; + LWLockRelease(IvmControlLock); +} + +/* + * getLastUpdateXid + * + * Get the most recent transaction ID that updated the view incrementally. + */ +static FullTransactionId +getLastUpdateXid(Oid immv_oid) +{ + FullTransactionId xid = InvalidFullTransactionId; + LastIvmUpdateEntry *entry; + bool found; + + LWLockAcquire(IvmControlLock, LW_SHARED); + entry = (LastIvmUpdateEntry *) hash_search(LastIvmUpdateHash, (void *) &immv_oid, HASH_FIND, &found); + LWLockRelease(IvmControlLock); + if (found) + xid = entry->last_ivm_update; + + return xid; +} + +void +removeImmv(Oid immv_oid) +{ + DroppedImmvInfo *entry; + + if (!dropped_immv) + mv_InitHashTables(); + + entry = (DroppedImmvInfo *) hash_search(dropped_immv, (void *) &immv_oid, HASH_ENTER, NULL); + entry->subxid = GetCurrentSubTransactionId(); +} + +/* + * isIvmName + * + * Check if this is a IVM hidden column from the name. + */ +bool +isIvmName(const char *s) +{ + if (s) + return (strncmp(s, "__ivm_", 6) == 0); + return false; +} diff --git a/src/backend/commands/tablecmds.c b/src/backend/commands/tablecmds.c index a1845240a98..99bc709959a 100644 --- a/src/backend/commands/tablecmds.c +++ b/src/backend/commands/tablecmds.c @@ -62,6 +62,7 @@ #include "commands/defrem.h" #include "commands/event_trigger.h" #include "commands/extension.h" +#include "commands/matview.h" #include "commands/repack.h" #include "commands/sequence.h" #include "commands/tablecmds.h" @@ -1804,6 +1805,9 @@ RangeVarCallbackForDropRelation(const RangeVar *rel, Oid relOid, Oid oldRelOid, state->actual_relkind = classform->relkind; state->actual_relpersistence = classform->relpersistence; + if (classform->relkind == RELKIND_MATVIEW && classform->relisivm) + removeImmv(relOid); + /* * Both RELKIND_RELATION and RELKIND_PARTITIONED_TABLE are OBJECT_TABLE, * but RemoveRelations() can only pass one relkind for a given relation. diff --git a/src/backend/utils/activity/wait_event_names.txt b/src/backend/utils/activity/wait_event_names.txt index 560659f9568..d97c35eeef0 100644 --- a/src/backend/utils/activity/wait_event_names.txt +++ b/src/backend/utils/activity/wait_event_names.txt @@ -370,6 +370,7 @@ WaitLSN "Waiting to read or update shared Wait-for-LSN state." LogicalDecodingControl "Waiting to read or update logical decoding status information." DataChecksumsWorker "Waiting for data checksums worker." AioWorkerControl "Waiting to update AIO worker information." +IvmControl "Waiting for read or update IMMV information.." # # END OF PREDEFINED LWLOCKS (DO NOT CHANGE THIS LINE) diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat index be157a5fbe9..b8722eea7a4 100644 --- a/src/include/catalog/pg_proc.dat +++ b/src/include/catalog/pg_proc.dat @@ -12693,4 +12693,14 @@ proname => 'hashoid8extended', prorettype => 'int8', proargtypes => 'oid8 int8', prosrc => 'hashoid8extended' }, +# IVM +{ oid => '786', descr => 'ivm trigger (before)', + proname => 'IVM_immediate_before', provolatile => 'v', prorettype => 'trigger', + proargtypes => '', prosrc => 'IVM_immediate_before' }, +{ oid => '787', descr => 'ivm trigger (after)', + proname => 'IVM_immediate_maintenance', provolatile => 'v', prorettype => 'trigger', + proargtypes => '', prosrc => 'IVM_immediate_maintenance' }, +{ oid => '788', descr => 'ivm filetring ', + proname => 'ivm_visible_in_prestate', provolatile => 's', prorettype => 'bool', + proargtypes => 'oid tid oid', prosrc => 'ivm_visible_in_prestate' }, ] diff --git a/src/include/commands/createas.h b/src/include/commands/createas.h index f895c67c0c2..c286ebcd70e 100644 --- a/src/include/commands/createas.h +++ b/src/include/commands/createas.h @@ -16,6 +16,7 @@ #include "catalog/objectaddress.h" #include "nodes/params.h" +#include "nodes/pathnodes.h" #include "parser/parse_node.h" #include "tcop/dest.h" #include "utils/queryenvironment.h" @@ -25,6 +26,9 @@ extern ObjectAddress ExecCreateTableAs(ParseState *pstate, CreateTableAsStmt *st ParamListInfo params, QueryEnvironment *queryEnv, QueryCompletion *qc); +extern void CreateIvmTriggersOnBaseTables(Query *qry, Oid matviewOid); +extern void CreateIndexOnIMMV(Query *query, Relation matviewRel); + extern int GetIntoRelEFlags(IntoClause *intoClause); extern DestReceiver *CreateIntoRelDestReceiver(IntoClause *intoClause); diff --git a/src/include/commands/matview.h b/src/include/commands/matview.h index 738c731c1a9..f1dbfc052c5 100644 --- a/src/include/commands/matview.h +++ b/src/include/commands/matview.h @@ -15,6 +15,7 @@ #define MATVIEW_H #include "catalog/objectaddress.h" +#include "fmgr.h" #include "nodes/params.h" #include "nodes/parsenodes.h" #include "tcop/dest.h" @@ -23,6 +24,8 @@ extern void SetMatViewPopulatedState(Relation relation, bool newstate); +extern void SetMatViewIVMState(Relation relation, bool newstate); + extern ObjectAddress ExecRefreshMatView(RefreshMatViewStmt *stmt, const char *queryString, QueryCompletion *qc); extern ObjectAddress RefreshMatViewByOid(Oid matviewOid, bool is_create, bool skipData, @@ -33,4 +36,12 @@ extern DestReceiver *CreateTransientRelDestReceiver(Oid transientoid); extern bool MatViewIncrementalMaintenanceIsEnabled(void); +extern Datum IVM_immediate_before(PG_FUNCTION_ARGS); +extern Datum IVM_immediate_maintenance(PG_FUNCTION_ARGS); +extern Datum IVM_visible_in_prestate(PG_FUNCTION_ARGS); +extern void AtAbort_IVM(SubTransactionId subtxid); +extern void AtPreCommit_IVM(void); +extern void removeImmv(Oid immv_oid); +extern bool isIvmName(const char *s); + #endif /* MATVIEW_H */ diff --git a/src/include/storage/lwlocklist.h b/src/include/storage/lwlocklist.h index d7eb648bd27..756c63baf97 100644 --- a/src/include/storage/lwlocklist.h +++ b/src/include/storage/lwlocklist.h @@ -89,6 +89,7 @@ PG_LWLOCK(54, WaitLSN) PG_LWLOCK(55, LogicalDecodingControl) PG_LWLOCK(56, DataChecksumsWorker) PG_LWLOCK(57, AioWorkerControl) +PG_LWLOCK(58, IvmControl) /* * There also exist several built-in LWLock tranches. As with the predefined diff --git a/src/include/storage/subsystemlist.h b/src/include/storage/subsystemlist.h index 9ad619080be..481e15d00a1 100644 --- a/src/include/storage/subsystemlist.h +++ b/src/include/storage/subsystemlist.h @@ -88,3 +88,6 @@ PG_SHMEM_SUBSYSTEM(DataChecksumsShmemCallbacks) /* AIO subsystem. This delegates to the method-specific callbacks */ PG_SHMEM_SUBSYSTEM(AioShmemCallbacks) + +/* Incremental View Maintenance */ +PG_SHMEM_SUBSYSTEM(IvmShmemCallbacks) -- 2.43.0