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

Add support for PostgreSQL 16 (#69) (#70)

Browse source 
Build errors/warnings against PostgreSQL 16 are fixed.
Also, adapted to the change of codes, including:

- Get rid of the "new" and "old" entries in a view's rangetable.
(Although, removed codes were dead codes because pg_ivm doesn't
 have any rules in pg_rewrite.)

- Rework query relation permission checking

- Require empty Bitmapsets to be represented as NULL

- Fix some compiler warnings
This commit is contained in:
Yugo Nagata 2023-09-11 15:23:51 +09:00 committed by GitHub
parent 6f99049848
commit 71f9d268b0
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
5 changed files with 83 additions and 72 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -78,7 +78,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);
@ -262,14 +262,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);
@ -297,7 +297,6 @@ rewriteQueryForIMMV(Query *query, List *colNames)
{
Query *rewritten;
TargetEntry *tle;
Node *node;
ParseState *pstate = make_parsestate(NULL);
FuncCall *fn;
@ -348,10 +347,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));
}
}
@ -399,6 +398,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
@ -516,22 +517,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;
/*
@ -553,10 +546,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);
@ -927,8 +919,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);
@ -1284,7 +1274,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;
@ -1294,14 +1284,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.
@ -1341,9 +1323,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;
@ -1396,7 +1383,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)
@ -1460,6 +1447,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,
@ -1500,7 +1490,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;
@ -1527,35 +1517,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;
@ -1567,14 +1549,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))
{
@ -1588,7 +1573,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);
}
}
@ -1596,7 +1586,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;
}

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;

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);