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