luciano/pg_ivm
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Merge branch 'main' into fix_exists

Browse source
This commit is contained in:
thoshiai 2024-01-15 02:46:06 +09:00
commit 6f5ae0ee22
9 changed files with 89 additions and 75 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
Makefile
View file

@ -13,7 +13,8 @@ PGFILEDESC = "pg_ivm - incremental view maintenance on PostgreSQL"
EXTENSION = pg_ivm
DATA = pg_ivm--1.0.sql \
pg_ivm--1.0--1.1.sql pg_ivm--1.1--1.2.sql pg_ivm--1.2--1.3.sql \
pg_ivm--1.3--1.4.sql pg_ivm--1.4--1.5.sql pg_ivm--1.5--1.6.sql
pg_ivm--1.3--1.4.sql pg_ivm--1.4--1.5.sql pg_ivm--1.5--1.6.sql \
pg_ivm--1.6--1.7.sql
REGRESS = pg_ivm create_immv refresh_immv

2
README.md
View file

@ -2,7 +2,7 @@
The `pg_ivm` module provides Incremental View Maintenance (IVM) feature for PostgreSQL.
The extension is compatible with PostgreSQL 13, 14, and 15.
The extension is compatible with PostgreSQL 13, 14, 15, and 16.
## Description

112
createas.c
View file

@ -79,7 +79,7 @@ static void CreateIvmTriggersOnBaseTablesRecurse(Query *qry, Node *node, Oid mat
static void CreateIvmTrigger(Oid relOid, Oid viewOid, int16 type, int16 timing, bool ex_lock);
static void check_ivm_restriction(Node *node);
static bool check_ivm_restriction_walker(Node *node, check_ivm_restriction_context *context);
static Bitmapset *get_primary_key_attnos_from_query(Query *query, List **constraintList, bool is_create);
static Bitmapset *get_primary_key_attnos_from_query(Query *query, List **constraintList);
static bool check_aggregate_supports_ivm(Oid aggfnoid);
static void StoreImmvQuery(Oid viewOid, bool ispopulated, Query *viewQuery);
@ -263,14 +263,14 @@ ExecCreateImmv(ParseState *pstate, CreateTableAsStmt *stmt,
if (!into->skipData)
{
/* Create an index on incremental maintainable materialized view, if possible */
CreateIndexOnIMMV(viewQuery, matviewRel, true);
CreateIndexOnIMMV(viewQuery, matviewRel);
/*
* Create triggers on incremental maintainable materialized view
* This argument should use 'query'. This needs to use a rewritten query,
* because a sublink in jointree is not supported by this function.
*/
CreateIvmTriggersOnBaseTables(query, matviewOid, true);
CreateIvmTriggersOnBaseTables(query, matviewOid);
/* Create triggers to prevent IMMV from beeing changed */
CreateChangePreventTrigger(matviewOid);
@ -298,7 +298,6 @@ rewriteQueryForIMMV(Query *query, List *colNames)
{
Query *rewritten;
TargetEntry *tle;
Node *node;
ParseState *pstate = make_parsestate(NULL);
FuncCall *fn;
@ -349,10 +348,10 @@ rewriteQueryForIMMV(Query *query, List *colNames)
if (countCol != NULL)
{
tle = makeTargetEntry((Expr *) countCol,
list_length(rewritten->targetList) + 1,
pstrdup(columnName),
false);
TargetEntry *tle = makeTargetEntry((Expr *) countCol,
list_length(rewritten->targetList) + 1,
pstrdup(columnName),
false);
rewritten->targetList = list_concat(rewritten->targetList, list_make1(tle));
}
}
@ -400,6 +399,8 @@ rewriteQueryForIMMV(Query *query, List *colNames)
/* Add count(*) for counting distinct tuples in views */
if (rewritten->distinctClause || rewritten->hasAggs)
{
TargetEntry *tle;
#if defined(PG_VERSION_NUM) && (PG_VERSION_NUM >= 140000)
fn = makeFuncCall(SystemFuncName("count"), NIL, COERCE_EXPLICIT_CALL, -1);
#else
@ -517,22 +518,14 @@ makeIvmAggColumn(ParseState *pstate, Aggref *aggref, char *resname, AttrNumber *
* CreateIvmTriggersOnBaseTables -- create IVM triggers on all base tables
*/
void
CreateIvmTriggersOnBaseTables(Query *qry, Oid matviewOid, bool is_create)
CreateIvmTriggersOnBaseTables(Query *qry, Oid matviewOid)
{
Relids relids = NULL;
bool ex_lock = false;
Index first_rtindex = is_create ? 1 : PRS2_NEW_VARNO + 1;
RangeTblEntry *rte;
/*
* is_create must be true in pg_ivm because the view definition doesn't
* contain NEW/OLD RTE.
* XXX: This argument should be removed?
*/
Assert(is_create);
/* Immediately return if we don't have any base tables. */
if (list_length(qry->rtable) < first_rtindex)
if (list_length(qry->rtable) < 1)
return;
/*
@ -554,10 +547,9 @@ CreateIvmTriggersOnBaseTables(Query *qry, Oid matviewOid, bool is_create)
* target list are not nullable.
*/
rte = list_nth(qry->rtable, first_rtindex - 1);
if (list_length(qry->rtable) > first_rtindex ||
rte->rtekind != RTE_RELATION || qry->distinctClause ||
(qry->hasAggs && qry->groupClause))
rte = list_nth(qry->rtable, 0);
if (list_length(qry->rtable) > 1 || rte->rtekind != RTE_RELATION ||
qry->distinctClause || (qry->hasAggs && qry->groupClause))
ex_lock = true;
CreateIvmTriggersOnBaseTablesRecurse(qry, (Node *)qry, matviewOid, &relids, ex_lock);
@ -930,8 +922,6 @@ check_ivm_restriction_walker(Node *node, check_ivm_restriction_context *context)
*/
if (context->exists_qual_vars != NIL && context->sublevels_up == 0)
{
ListCell *lc;
foreach (lc, context->exists_qual_vars)
{
Var *var = (Var *) lfirst(lc);
@ -1300,7 +1290,7 @@ check_aggregate_supports_ivm(Oid aggfnoid)
* is created on these attritubes. In other cases, no index is created.
*/
void
CreateIndexOnIMMV(Query *query, Relation matviewRel, bool is_create)
CreateIndexOnIMMV(Query *query, Relation matviewRel)
{
ListCell *lc;
IndexStmt *index;
@ -1310,14 +1300,6 @@ CreateIndexOnIMMV(Query *query, Relation matviewRel, bool is_create)
List *indexoidlist = RelationGetIndexList(matviewRel);
ListCell *indexoidscan;
/*
* is_create must be true in pg_ivm because the view definition doesn't
* contain NEW/OLD RTE.
* XXX: This argument should be removed?
*/
Assert(is_create);
/*
* For aggregate without GROUP BY, we do not need to create an index
* because the view has only one row.
@ -1357,9 +1339,14 @@ CreateIndexOnIMMV(Query *query, Relation matviewRel, bool is_create)
index->excludeOpNames = NIL;
index->idxcomment = NULL;
index->indexOid = InvalidOid;
#if defined(PG_VERSION_NUM) && (PG_VERSION_NUM >= 160000)
index->oldNumber = InvalidRelFileNumber;
index->oldFirstRelfilelocatorSubid = InvalidSubTransactionId;
#else
index->oldNode = InvalidOid;
index->oldCreateSubid = InvalidSubTransactionId;
index->oldFirstRelfilenodeSubid = InvalidSubTransactionId;
#endif
index->oldCreateSubid = InvalidSubTransactionId;
index->transformed = true;
index->concurrent = false;
index->if_not_exists = false;
@ -1412,7 +1399,7 @@ CreateIndexOnIMMV(Query *query, Relation matviewRel, bool is_create)
Bitmapset *key_attnos;
/* create index on the base tables' primary key columns */
key_attnos = get_primary_key_attnos_from_query(query, &constraintList, is_create);
key_attnos = get_primary_key_attnos_from_query(query, &constraintList);
if (key_attnos)
{
foreach(lc, query->targetList)
@ -1476,6 +1463,9 @@ CreateIndexOnIMMV(Query *query, Relation matviewRel, bool is_create)
InvalidOid,
InvalidOid,
InvalidOid,
#if defined(PG_VERSION_NUM) && (PG_VERSION_NUM >= 160000)
-1,
#endif
false, true, false, false, true);
ereport(NOTICE,
@ -1516,7 +1506,7 @@ CreateIndexOnIMMV(Query *query, Relation matviewRel, bool is_create)
* constraintList is set to a list of the OIDs of the pkey constraints.
*/
static Bitmapset *
get_primary_key_attnos_from_query(Query *query, List **constraintList, bool is_create)
get_primary_key_attnos_from_query(Query *query, List **constraintList)
{
List *key_attnos_list = NIL;
ListCell *lc;
@ -1543,35 +1533,27 @@ get_primary_key_attnos_from_query(Query *query, List **constraintList, bool is_c
* Collect primary key attributes from all tables used in query. The key attributes
* sets for each table are stored in key_attnos_list in order by RTE index.
*/
i = 1;
foreach(lc, query->rtable)
{
RangeTblEntry *r = (RangeTblEntry*) lfirst(lc);
Bitmapset *key_attnos;
bool has_pkey = true;
Index first_rtindex = is_create ? 1 : PRS2_NEW_VARNO + 1;
/* skip NEW/OLD entries */
if (i >= first_rtindex)
/* for subqueries, scan recursively */
if (r->rtekind == RTE_SUBQUERY)
{
/* for subqueries, scan recursively */
if (r->rtekind == RTE_SUBQUERY)
{
key_attnos = get_primary_key_attnos_from_query(r->subquery, constraintList, true);
has_pkey = (key_attnos != NULL);
}
/* for tables, call get_primary_key_attnos */
else if (r->rtekind == RTE_RELATION)
{
Oid constraintOid;
key_attnos = get_primary_key_attnos(r->relid, false, &constraintOid);
*constraintList = lappend_oid(*constraintList, constraintOid);
has_pkey = (key_attnos != NULL);
}
/* for other RTEs, store NULL into key_attnos_list */
else
key_attnos = NULL;
key_attnos = get_primary_key_attnos_from_query(r->subquery, constraintList);
has_pkey = (key_attnos != NULL);
}
/* for tables, call get_primary_key_attnos */
else if (r->rtekind == RTE_RELATION)
{
Oid constraintOid;
key_attnos = get_primary_key_attnos(r->relid, false, &constraintOid);
*constraintList = lappend_oid(*constraintList, constraintOid);
has_pkey = (key_attnos != NULL);
}
/* for other RTEs, store NULL into key_attnos_list */
else
key_attnos = NULL;
@ -1583,14 +1565,17 @@ get_primary_key_attnos_from_query(Query *query, List **constraintList, bool is_c
return NULL;
key_attnos_list = lappend(key_attnos_list, key_attnos);
i++;
}
/* Collect key attributes appearing in the target list */
i = 1;
foreach(lc, query->targetList)
{
#if defined(PG_VERSION_NUM) && (PG_VERSION_NUM >= 160000)
TargetEntry *tle = (TargetEntry *) flatten_join_alias_vars(NULL, query, lfirst(lc));
#else
TargetEntry *tle = (TargetEntry *) flatten_join_alias_vars(query, lfirst(lc));
#endif
if (IsA(tle->expr, Var))
{
@ -1604,7 +1589,12 @@ get_primary_key_attnos_from_query(Query *query, List **constraintList, bool is_c
* Remove found key attributes from key_attnos_list, and add this
* to the result list.
*/
bms_del_member(key_attnos, var->varattno - FirstLowInvalidHeapAttributeNumber);
key_attnos = bms_del_member(key_attnos, var->varattno - FirstLowInvalidHeapAttributeNumber);
if (bms_is_empty(key_attnos))
{
key_attnos_list = list_delete_nth_cell(key_attnos_list, var->varno - 1);
key_attnos_list = list_insert_nth(key_attnos_list, var->varno - 1, NULL);
}
keys = bms_add_member(keys, i - FirstLowInvalidHeapAttributeNumber);
}
}
@ -1612,7 +1602,11 @@ get_primary_key_attnos_from_query(Query *query, List **constraintList, bool is_c
}
/* Collect RTE indexes of relations appearing in the FROM clause */
#if defined(PG_VERSION_NUM) && (PG_VERSION_NUM >= 160000)
rels_in_from = get_relids_in_jointree((Node *) query->jointree, false, false);
#else
rels_in_from = get_relids_in_jointree((Node *) query->jointree, false);
#endif
/*
* Check if all key attributes of relations in FROM are appearing in the target

33
matview.c
View file

@ -354,9 +354,6 @@ ExecRefreshImmv(const RangeVar *relation, bool skipData,
{
Relation tgRel;
Relation depRel;
ScanKeyData key;
SysScanDesc scan;
HeapTuple tup;
ObjectAddresses *immv_triggers;
immv_triggers = new_object_addresses();
@ -450,7 +447,7 @@ ExecRefreshImmv(const RangeVar *relation, bool skipData,
pgstat_count_heap_insert(matviewRel, processed);
if (!skipData && !oldPopulated)
CreateIvmTriggersOnBaseTables(viewQuery, matviewOid, true);
CreateIvmTriggersOnBaseTables(viewQuery, matviewOid);
table_close(matviewRel, NoLock);
@ -905,8 +902,13 @@ IVM_immediate_maintenance(PG_FUNCTION_ARGS)
if (!(query->hasAggs && query->groupClause == NIL))
{
OpenImmvIncrementalMaintenance();
#if defined(PG_VERSION_NUM) && (PG_VERSION_NUM >= 160000)
ExecuteTruncateGuts(list_make1(matviewRel), list_make1_oid(matviewOid),
NIL, DROP_RESTRICT, false, false);
#else
ExecuteTruncateGuts(list_make1(matviewRel), list_make1_oid(matviewOid),
NIL, DROP_RESTRICT, false);
#endif
CloseImmvIncrementalMaintenance();
}
else
@ -1041,7 +1043,6 @@ IVM_immediate_maintenance(PG_FUNCTION_ARGS)
foreach(lc2, table->rte_paths)
{
List *rte_path = lfirst(lc2);
int i;
Query *querytree = rewritten;
RangeTblEntry *rte;
TupleDesc tupdesc_old;
@ -1359,7 +1360,7 @@ get_prestate_rte(RangeTblEntry *rte, MV_TriggerTable *table,
{
StringInfoData str;
RawStmt *raw;
Query *sub;
Query *subquery;
Relation rel;
ParseState *pstate;
char *relname;
@ -1371,7 +1372,7 @@ get_prestate_rte(RangeTblEntry *rte, MV_TriggerTable *table,
/*
* We can use NoLock here since AcquireRewriteLocks should
* have locked the rel already.
* have locked the relation already.
*/
rel = table_open(table->table_id, NoLock);
relname = quote_qualified_identifier(
@ -1379,12 +1380,19 @@ get_prestate_rte(RangeTblEntry *rte, MV_TriggerTable *table,
RelationGetRelationName(rel));
table_close(rel, NoLock);
/*
* Filtering inserted row using the snapshot taken before the table
* is modified. ctid is required for maintaining outer join views.
*/
initStringInfo(&str);
appendStringInfo(&str,
"SELECT t.* FROM %s t"
" WHERE pg_catalog.ivm_visible_in_prestate(t.tableoid, t.ctid ,%d::pg_catalog.oid)",
relname, matviewid);
/*
* Append deleted rows contained in old transition tables.
*/
for (i = 0; i < list_length(table->old_tuplestores); i++)
{
appendStringInfo(&str, " UNION ALL ");
@ -1392,20 +1400,21 @@ get_prestate_rte(RangeTblEntry *rte, MV_TriggerTable *table,
make_delta_enr_name("old", table->table_id, i));
}
/* Get a subquery representing pre-state of the table */
#if defined(PG_VERSION_NUM) && (PG_VERSION_NUM >= 140000)
raw = (RawStmt*)linitial(raw_parser(str.data, RAW_PARSE_DEFAULT));
#else
raw = (RawStmt*)linitial(raw_parser(str.data));
#endif
sub = transformStmt(pstate, raw->stmt);
subquery = transformStmt(pstate, raw->stmt);
/* save the original RTE */
table->original_rte = copyObject(rte);
rte->rtekind = RTE_SUBQUERY;
rte->subquery = sub;
rte->subquery = subquery;
rte->security_barrier = false;
/* Clear fields that should not be set in a subquery RTE */
rte->relid = InvalidOid;
rte->relkind = 0;
@ -1413,12 +1422,16 @@ get_prestate_rte(RangeTblEntry *rte, MV_TriggerTable *table,
rte->tablesample = NULL;
rte->inh = false; /* must not be set for a subquery */
#if defined(PG_VERSION_NUM) && (PG_VERSION_NUM >= 160000)
rte->perminfoindex = 0; /* no permission checking for this RTE */
#else
rte->requiredPerms = 0; /* no permission check on subquery itself */
rte->checkAsUser = InvalidOid;
rte->selectedCols = NULL;
rte->insertedCols = NULL;
rte->updatedCols = NULL;
rte->extraUpdatedCols = NULL;
#endif
return rte;
}

0
pg_ivm--1.6--1.7.sql Normal file
View file

4
pg_ivm.c
View file

@ -122,7 +122,11 @@ parseNameAndColumns(const char *string, List **names, List **colNames)
/* Separate the name and parse it into a list */
*ptr++ = '\0';
#if defined(PG_VERSION_NUM) && (PG_VERSION_NUM >= 160000)
*names = stringToQualifiedNameList(rawname, NULL);
#else
*names = stringToQualifiedNameList(rawname);
#endif
if (!has_colnames)
goto end;

2
pg_ivm.control
View file

@ -1,6 +1,6 @@
# incremental view maintenance extension_
comment = 'incremental view maintenance on PostgreSQL'
default_version = '1.6'
default_version = '1.7'
module_pathname = '$libdir/pg_ivm'
relocatable = false
schema = pg_catalog

4
pg_ivm.h
View file

@ -38,8 +38,8 @@ extern bool isImmv(Oid immv_oid);
extern ObjectAddress ExecCreateImmv(ParseState *pstate, CreateTableAsStmt *stmt,
ParamListInfo params, QueryEnvironment *queryEnv,
QueryCompletion *qc);
extern void CreateIvmTriggersOnBaseTables(Query *qry, Oid matviewOid, bool is_create);
extern void CreateIndexOnIMMV(Query *query, Relation matviewRel, bool is_create);
extern void CreateIvmTriggersOnBaseTables(Query *qry, Oid matviewOid);
extern void CreateIndexOnIMMV(Query *query, Relation matviewRel);
extern Query *rewriteQueryForIMMV(Query *query, List *colNames);
extern void makeIvmAggColumn(ParseState *pstate, Aggref *aggref, char *resname, AttrNumber *next_resno, List **aggs);

4
rpm/pg_ivm.spec
View file

@ -10,7 +10,7 @@
Summary: PostgreSQL-based distributed RDBMS
Name: %{sname}_%{pgmajorversion}
Version: 1.6
Version: 1.7
Release: 1%{dist}
License: BSD
Vendor: IVM Development Group
@ -55,6 +55,8 @@ PATH=%{pginstdir}/bin:$PATH %{__make} %{?_smp_mflags} INSTALL_PREFIX=%{buildroot
%endif
%changelog
* Thu Sep 13 2023 - Yugo Nagata <nagata@sraoss.co.jp> 1.7-1
- Update to 1.7
* Thu Aug 31 2023 - Yugo Nagata <nagata@sraoss.co.jp> 1.6-1
- Update to 1.6
* Thu Mar 2 2023 - Yugo Nagata <nagata@sraoss.co.jp> 1.5.1-1