前面一篇文档主要讲了增量物化视图的使用，这里主要讲解一下其设计原理，设计原理可参考Incremental View Maintenance，里面说的非常明白。
基础理论
IVM computes and applies only the incremental changes to the materialized views. Suppose that view V is defined by query Q over a state of base relations D. When D changes D’ = D + dD, we can get the new view state V’ by calculating from D’ and Q, and this is re-computation performed by REFRESH MATERIALIZED VIEW command. On the other hand, IVM calculates the delta for view (dV) from the base tables delta (dD) and view definition (Q), and applies this to get the new view state, V’ = V + dV.
In theory, the view definition is described in a relational algebra (or bag algebra) form. For example, a (inner) join view of table R and S is defined as V = R ⨝ S.
When table R is changed in a transaction, this can be described as R ← R - ∇R + ΔR, where ∇R and ΔR denote tuples deleted from and inserted into R, respectively. (To be accurate, instead of - and +, ∸ and ⨄ are used by tradition in bag algebra.) In this condition, the deltas of the view are calculated as ∇V = ∇R ⨝ S and ΔV = ΔR ⨝ S, then the view can be updated as V ← V - ∇V + ΔV.
术语：
- 基表（Base table）:指物化视图定义中使用到的普通表
- 增量（Delta）：指基表的数据发生变化时，与物化视图中的数据相比，增加和删除的数据集合。
如何监测基表的改动呢？
核心是触发器（after trigger）和转换表机制， 是基于触发器去实现的，在基表上创建INSERT | UPDAE | DELETE | TRUNCATE AFTER STATEMENT触发器，当对基表修改时，通过触发器执行触发器函数，触发器中有个特性，支持REFERENCING选项。
REFERENCING选项启用对传递关系的收集，传递关系是包括被当前SQL语句插入、删除或者修改的行的行集合。这个特性让触发器能看到该语句做的事情的全局视图，而不是一次只看到一行。仅对非约束触发器的AFTER触发器允许这个选项。此外，如果触发器是一个UPDATE触发器，则它不能指定column_name列表。OLD TABLE仅可以被指定一次，并且只能为在UPDATE或DELETE事件上引发的触发器指定，它创建的传递关系包含有该语句更新或删除的所有行的前映像。类似地，NEW TABLE仅可以被指定一次，并且只能为在UPDATE或INSERT事件上引发的触发器指定，它创建的传递关系包含有该语句更新或插入的所有行的后映像。
CREATE [ OR REPLACE ] [ CONSTRAINT ] TRIGGER name { BEFORE | AFTER | INSTEAD OF } { event [ OR ... ] }
ON table_name
[ FROM referenced_table_name ]
[ NOT DEFERRABLE | [ DEFERRABLE ] [ INITIALLY IMMEDIATE | INITIALLY DEFERRED ] ]
[ REFERENCING { { OLD | NEW } TABLE [ AS ] transition_relation_name } [ ... ] ]
[ FOR [ EACH ] { ROW | STATEMENT } ]
[ WHEN ( condition ) ]
EXECUTE { FUNCTION | PROCEDURE } function_name ( arguments )
这里的event可以是下列之一：
INSERT
UPDATE [ OF column_name [, ... ] ]
DELETE
TRUNCATE
如何根据基表的增量计算物化视图的增量？
前面已通过触发器以及转换表机制，已经实现了对基表的改动的监测以及获取了增量数据，那么如何根据基表的增量计算物化视图的增量呢？
-- view definition
select ... fom A,B where ...
-- rewrite query
-- 这里new_table_A，就是存储的基表A新增的数据
-- 经过计算就是物化视图新增的数据
select ... from new_table_A,B where ...
-- 这里old_table_A，就是存储的基表A删除的数据
-- 经过计算就是物化视图删除的数据
select ... from old_table_A,B where ...
怎么把增量数据应用到物化视图中呢？
实质就是在计算出物化视图的增量数据（分为需要insert的以及delete的）后，对物化视图表进行insert和delete操作。这也是为什么在增量物化视图的创建过程中，会根据视图定义自动创建增量物化视图表的索引，创建索引后，对增量物化视图表的更新操作就会更加高效。
部分源码解析
核心实现函数：当基表发生更新时，会触发After触发器，调用该函数。
/*
* IVM_immediate_maintenance
*
* IVM trigger function invoked after base table is modified.
* For each table, tuplestores of transition tables are collected.
* and after the last modification
*/
Datum IVM_immediate_maintenance(PG_FUNCTION_ARGS)
{
MV_TriggerTable *table;
/* Save the transition tables and make a request to not free immediately */
// 获取增量数据，delete数据
if (trigdata->tg_oldtable)
{
oldcxt = MemoryContextSwitchTo(TopTransactionContext);
table->old_tuplestores = lappend(table->old_tuplestores, tuplestore_copy(trigdata->tg_oldtable, rel));
entry->has_old = true;
MemoryContextSwitchTo(oldcxt);
}
// 获取增量数据，insert数据
if (trigdata->tg_newtable)
{
oldcxt = MemoryContextSwitchTo(TopTransactionContext);
table->new_tuplestores = lappend(table->new_tuplestores, tuplestore_copy(trigdata->tg_newtable, rel));
entry->has_new = true;
MemoryContextSwitchTo(oldcxt);
}
/* calculate delta tables */
// 计算增量数据，根据基表的oldtable,newtable 应用到视图定义中得到视图的增量数据
calc_delta(table, rte_path, rewritten, dest_old, dest_new, &tupdesc_old, &tupdesc_new, queryEnv);
/* apply the delta tables to the materialized view */
// 将视图增量数据应用到物化视图表中，也就是将上一步的计算结果，通过insert/delete更新物化视图
apply_delta(matviewOid, old_tuplestore, new_tuplestore,tupdesc_old, tupdesc_new, query, use_count,count_colname);
}
/*
* apply_new_delta
*
* Execute a query for applying a delta table given by deltaname_new
* which contains tuples to be inserted into a materialized view given by
* matviewname. This is used when counting is not required.
*/
static void
apply_new_delta(const char *matviewname, const char *deltaname_new,
StringInfo target_list)
{
StringInfoData querybuf;
/* Search for matching tuples from the view and update or delete if found. */
initStringInfo(&querybuf);
appendStringInfo(&querybuf,
"INSERT INTO %s (%s) SELECT %s FROM ("
"SELECT diff.*, pg_catalog.generate_series(1, diff.\"__ivm_count__\") AS __ivm_generate_series__ "
"FROM %s AS diff) AS v",
matviewname, target_list->data, target_list->data,
deltaname_new);
if (SPI_exec(querybuf.data, 0) != SPI_OK_INSERT)
elog(ERROR, "SPI_exec failed: %s", querybuf.data);
}
/*
* apply_old_delta
*
* Execute a query for applying a delta table given by deltaname_old
* which contains tuples to be deleted from a materialized view given by
* matviewname. This is used when counting is not required.
*/
static void
apply_old_delta(const char *matviewname, const char *deltaname_old,
List *keys)
{
StringInfoData querybuf;
StringInfoData keysbuf;
char *match_cond;
ListCell *lc;
/* build WHERE condition for searching tuples to be deleted */
match_cond = get_matching_condition_string(keys);
/* build string of keys list */
initStringInfo(&keysbuf);
foreach (lc, keys)
{
Form_pg_attribute attr = (Form_pg_attribute) lfirst(lc);
char *resname = NameStr(attr->attname);
appendStringInfo(&keysbuf, "%s", quote_qualified_identifier("mv", resname));
if (lnext(keys, lc))
appendStringInfo(&keysbuf, ", ");
}
/* Search for matching tuples from the view and update or delete if found. */
initStringInfo(&querybuf);
appendStringInfo(&querybuf,
"DELETE FROM %s WHERE ctid IN ("
"SELECT tid FROM (SELECT pg_catalog.row_number() over (partition by %s) AS \"__ivm_row_number__\","
"mv.ctid AS tid,"
"diff.\"__ivm_count__\""
"FROM %s AS mv, %s AS diff "
"WHERE %s) v "
"WHERE v.\"__ivm_row_number__\" OPERATOR(pg_catalog.<=) v.\"__ivm_count__\")",
matviewname,
keysbuf.data,
matviewname, deltaname_old,
match_cond);
if (SPI_exec(querybuf.data, 0) != SPI_OK_DELETE)
elog(ERROR, "SPI_exec failed: %s", querybuf.data);
}
