public inbox for [email protected]help / color / mirror / Atom feed
[PATCH 1/2] index_get_partition 11+ messages / 6 participants [nested] [flat]
* [PATCH 1/2] index_get_partition @ 2019-01-22 21:00 Alvaro Herrera <[email protected]> 0 siblings, 0 replies; 11+ messages in thread From: Alvaro Herrera @ 2019-01-22 21:00 UTC (permalink / raw) --- src/backend/catalog/partition.c | 35 +++++++++++++++++++++++++++++++++++ src/backend/commands/tablecmds.c | 40 +++++++++++----------------------------- src/include/catalog/partition.h | 1 + 3 files changed, 47 insertions(+), 29 deletions(-) diff --git a/src/backend/catalog/partition.c b/src/backend/catalog/partition.c index 0d3bc3a2c7..66012bb28a 100644 --- a/src/backend/catalog/partition.c +++ b/src/backend/catalog/partition.c @@ -146,6 +146,41 @@ get_partition_ancestors_worker(Relation inhRel, Oid relid, List **ancestors) } /* + * Return the OID of the index, in the given partition, that is a child of the + * given index or InvalidOid if there isn't one. + */ +Oid +index_get_partition(Relation partition, Oid indexId) +{ + List *idxlist = RelationGetIndexList(partition); + ListCell *l; + + foreach(l, idxlist) + { + Oid partIdx = lfirst_oid(l); + HeapTuple tup; + Form_pg_class classForm; + bool ispartition; + + tup = SearchSysCache1(RELOID, ObjectIdGetDatum(partIdx)); + if (!tup) + elog(ERROR, "cache lookup failed for relation %u", partIdx); + classForm = (Form_pg_class) GETSTRUCT(tup); + ispartition = classForm->relispartition; + ReleaseSysCache(tup); + if (!ispartition) + continue; + if (get_partition_parent(lfirst_oid(l)) == indexId) + { + list_free(idxlist); + return partIdx; + } + } + + return InvalidOid; +} + +/* * map_partition_varattnos - maps varattno of any Vars in expr from the * attno's of 'from_rel' to the attno's of 'to_rel' partition, each of which * may be either a leaf partition or a partitioned table, but both of which diff --git a/src/backend/commands/tablecmds.c b/src/backend/commands/tablecmds.c index 35a9ade059..877bac506f 100644 --- a/src/backend/commands/tablecmds.c +++ b/src/backend/commands/tablecmds.c @@ -15427,36 +15427,18 @@ ATExecAttachPartitionIdx(List **wqueue, Relation parentIdx, RangeVar *name) static void refuseDupeIndexAttach(Relation parentIdx, Relation partIdx, Relation partitionTbl) { - Relation pg_inherits; - ScanKeyData key; - HeapTuple tuple; - SysScanDesc scan; + Oid existingIdx; - pg_inherits = table_open(InheritsRelationId, AccessShareLock); - ScanKeyInit(&key, Anum_pg_inherits_inhparent, - BTEqualStrategyNumber, F_OIDEQ, - ObjectIdGetDatum(RelationGetRelid(parentIdx))); - scan = systable_beginscan(pg_inherits, InheritsParentIndexId, true, - NULL, 1, &key); - while (HeapTupleIsValid(tuple = systable_getnext(scan))) - { - Form_pg_inherits inhForm; - Oid tab; - - inhForm = (Form_pg_inherits) GETSTRUCT(tuple); - tab = IndexGetRelation(inhForm->inhrelid, false); - if (tab == RelationGetRelid(partitionTbl)) - ereport(ERROR, - (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), - errmsg("cannot attach index \"%s\" as a partition of index \"%s\"", - RelationGetRelationName(partIdx), - RelationGetRelationName(parentIdx)), - errdetail("Another index is already attached for partition \"%s\".", - RelationGetRelationName(partitionTbl)))); - } - - systable_endscan(scan); - table_close(pg_inherits, AccessShareLock); + existingIdx = index_get_partition(partitionTbl, + RelationGetRelid(parentIdx)); + if (OidIsValid(existingIdx)) + ereport(ERROR, + (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("cannot attach index \"%s\" as a partition of index \"%s\"", + RelationGetRelationName(partIdx), + RelationGetRelationName(parentIdx)), + errdetail("Another index is already attached for partition \"%s\".", + RelationGetRelationName(partitionTbl)))); } /* diff --git a/src/include/catalog/partition.h b/src/include/catalog/partition.h index 5685d2fd57..7f0b198bcb 100644 --- a/src/include/catalog/partition.h +++ b/src/include/catalog/partition.h @@ -35,6 +35,7 @@ typedef struct PartitionDescData extern Oid get_partition_parent(Oid relid); extern List *get_partition_ancestors(Oid relid); +extern Oid index_get_partition(Relation partition, Oid indexId); extern List *map_partition_varattnos(List *expr, int fromrel_varno, Relation to_rel, Relation from_rel, bool *found_whole_row); -- 2.11.0 --gy7na3sjffpqghel Content-Type: text/x-diff; charset=us-ascii Content-Disposition: attachment; filename="0002-Propagate-replica-identity-to-partitions.patch" ^ permalink raw reply [nested|flat] 11+ messages in thread
* [PATCH v30 08/11] Add aggregates support in IVM @ 2023-05-31 11:46 Yugo Nagata <[email protected]> 0 siblings, 0 replies; 11+ messages in thread From: Yugo Nagata @ 2023-05-31 11:46 UTC (permalink / raw) count, sum, adn avg are supported. As a restriction, expressions specified in GROUP BY must appear in the target list because tuples to be updated in IMMV are identified by using this group key. However, in the case of aggregates without GROUP BY, there is only one tuple in the view, so keys are not uses to identify tuples. When creating a IMMV, in addition to __ivm_count column, some hidden columns for each aggregate are added to the target list. For example, names of these hidden columns are ivm_count_avg and ivm_sum_avg for the average function, and so on. When a base table is modified, the aggregated values and related hidden columns are also updated as well as __ivm_count__. The way of update depends the kind of aggregate function.=E3=80=80Specifically, sum and count are updated by simply adding or subtracting delta value calculated from delta tables. avg is updated by using values of sum and count stored in views as hidden columns and deltas calculated from delta tables. About aggregate functions except "count()" (sum and avg), NULLs in input values are ignored, and the result of aggegate should be NULL when no rows are selected. To support this specification, the numbers of non-NULL input values are counted and stored in hidden columns. In the case of count(), count(x) returns zero when no rows are selected, but count(*) doesn't ignore NULL input. --- src/backend/commands/createas.c | 264 +++++++++++++++++-- src/backend/commands/matview.c | 433 ++++++++++++++++++++++++++++++-- src/include/commands/createas.h | 1 + 3 files changed, 661 insertions(+), 37 deletions(-) diff --git a/src/backend/commands/createas.c b/src/backend/commands/createa= s.c index cce44278fa..d93eec3eec 100644 --- a/src/backend/commands/createas.c +++ b/src/backend/commands/createas.c @@ -54,14 +54,19 @@ #include "parser/parsetree.h" #include "parser/parse_clause.h" #include "parser/parse_func.h" +#include "parser/parse_type.h" #include "rewrite/rewriteHandler.h" +#include "rewrite/rewriteManip.h" #include "storage/smgr.h" #include "tcop/tcopprot.h" #include "utils/builtins.h" #include "utils/lsyscache.h" +#include "utils/regproc.h" +#include "utils/fmgroids.h" #include "utils/rel.h" #include "utils/rls.h" #include "utils/snapmgr.h" +#include "utils/syscache.h" =20 typedef struct { @@ -75,6 +80,11 @@ typedef struct BulkInsertState bistate; /* bulk insert state */ } DR_intorel; =20 +typedef struct +{ + bool has_agg; +} check_ivm_restriction_context; + /* utility functions for CTAS definition creation */ static ObjectAddress create_ctas_internal(List *attrList, IntoClause *into= ); static ObjectAddress create_ctas_nodata(List *tlist, IntoClause *into); @@ -89,8 +99,9 @@ static void CreateIvmTriggersOnBaseTablesRecurse(Query *q= ry, Node *node, Oid mat Relids *relids, bool ex_lock); static void CreateIvmTrigger(Oid relOid, Oid viewOid, int16 type, int16 ti= ming, bool ex_lock); static void check_ivm_restriction(Node *node); -static bool check_ivm_restriction_walker(Node *node, void *context); +static bool check_ivm_restriction_walker(Node *node, check_ivm_restriction= _context *context); static Bitmapset *get_primary_key_attnos_from_query(Query *query, List **c= onstraintList); +static bool check_aggregate_supports_ivm(Oid aggfnoid); =20 /* * create_ctas_internal @@ -421,6 +432,7 @@ ExecCreateTableAs(ParseState *pstate, CreateTableAsStmt= *stmt, * rewriteQueryForIMMV -- rewrite view definition query for IMMV * * count(*) is added for counting distinct tuples in views. + * Also, additional hidden columns are added for aggregate values. */ Query * rewriteQueryForIMMV(Query *query, List *colNames) @@ -434,16 +446,49 @@ rewriteQueryForIMMV(Query *query, List *colNames) rewritten =3D copyObject(query); pstate->p_expr_kind =3D EXPR_KIND_SELECT_TARGET; =20 - /* - * Convert DISTINCT to GROUP BY and add count(*) for counting distinct - * tuples in views. - */ - if (rewritten->distinctClause) + /* group keys must be in targetlist */ + if (rewritten->groupClause) { - TargetEntry *tle; + ListCell *lc; + foreach(lc, rewritten->groupClause) + { + SortGroupClause *scl =3D (SortGroupClause *) lfirst(lc); + TargetEntry *tle =3D get_sortgroupclause_tle(scl, rewritten->targetList= ); =20 + if (tle->resjunk) + ereport(ERROR, + (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg("GROUP BY expression not appearing in select list is not sup= ported on incrementally maintainable materialized view"))); + } + } + /* Convert DISTINCT to GROUP BY. count(*) will be added afterward. */ + else if (!rewritten->hasAggs && rewritten->distinctClause) rewritten->groupClause =3D transformDistinctClause(NULL, &rewritten->tar= getList, rewritten->sortClause, false); =20 + /* Add additional columns for aggregate values */ + if (rewritten->hasAggs) + { + ListCell *lc; + List *aggs =3D NIL; + AttrNumber next_resno =3D list_length(rewritten->targetList) + 1; + + foreach(lc, rewritten->targetList) + { + TargetEntry *tle =3D (TargetEntry *) lfirst(lc); + char *resname =3D (colNames =3D=3D NIL || foreach_current_index(lc) >= =3D list_length(colNames) ? + tle->resname : strVal(list_nth(colNames, tle->resno - 1))); + + if (IsA(tle->expr, Aggref)) + makeIvmAggColumn(pstate, (Aggref *) tle->expr, resname, &next_resno, &= aggs); + } + rewritten->targetList =3D list_concat(rewritten->targetList, aggs); + } + + /* Add count(*) for counting distinct tuples in views */ + if (rewritten->distinctClause || rewritten->hasAggs) + { + TargetEntry *tle; + fn =3D makeFuncCall(SystemFuncName("count"), NIL, COERCE_EXPLICIT_CALL, = -1); fn->agg_star =3D true; =20 @@ -460,6 +505,91 @@ rewriteQueryForIMMV(Query *query, List *colNames) return rewritten; } =20 +/* + * makeIvmAggColumn -- make additional aggregate columns for IVM + * + * For an aggregate column specified by aggref, additional aggregate colum= ns + * are added, which are used to calculate the new aggregate value in IMMV. + * An additional aggregate columns has a name based on resname + * (ex. ivm_count_resname), and resno specified by next_resno. The created + * columns are returned to aggs, and the resno for the next column is also + * returned to next_resno. + * + * Currently, an additional count() is created for aggref other than count. + * In addition, sum() is created for avg aggregate column. + */ +void +makeIvmAggColumn(ParseState *pstate, Aggref *aggref, char *resname, AttrNu= mber *next_resno, List **aggs) +{ + TargetEntry *tle_count; + Node *node; + FuncCall *fn; + Const *dmy_arg =3D makeConst(INT4OID, + -1, + InvalidOid, + sizeof(int32), + Int32GetDatum(1), + false, + true); /* pass by value */ + const char *aggname =3D get_func_name(aggref->aggfnoid); + + /* + * For aggregate functions except count, add count() func with the same a= rg parameters. + * This count result is used for determining if the aggregate value shoul= d be NULL or not. + * Also, add sum() func for avg because we need to calculate an average v= alue as sum/count. + * + * XXX: If there are same expressions explicitly in the target list, we c= an use this instead + * of adding new duplicated one. + */ + if (strcmp(aggname, "count") !=3D 0) + { + fn =3D makeFuncCall(SystemFuncName("count"), NIL, COERCE_EXPLICIT_CALL, = -1); + + /* Make a Func with a dummy arg, and then override this by the original = agg's args. */ + node =3D ParseFuncOrColumn(pstate, fn->funcname, list_make1(dmy_arg), NU= LL, fn, false, -1); + ((Aggref *)node)->args =3D aggref->args; + + tle_count =3D makeTargetEntry((Expr *) node, + *next_resno, + pstrdup(makeObjectName("__ivm_count",resname, "_")), + false); + *aggs =3D lappend(*aggs, tle_count); + (*next_resno)++; + } + if (strcmp(aggname, "avg") =3D=3D 0) + { + List *dmy_args =3D NIL; + ListCell *lc; + foreach(lc, aggref->aggargtypes) + { + Oid typeid =3D lfirst_oid(lc); + Type type =3D typeidType(typeid); + + Const *con =3D makeConst(typeid, + -1, + typeTypeCollation(type), + typeLen(type), + (Datum) 0, + true, + typeByVal(type)); + dmy_args =3D lappend(dmy_args, con); + ReleaseSysCache(type); + } + fn =3D makeFuncCall(SystemFuncName("sum"), NIL, COERCE_EXPLICIT_CALL, -1= ); + + /* Make a Func with dummy args, and then override this by the original a= gg's args. */ + node =3D ParseFuncOrColumn(pstate, fn->funcname, dmy_args, NULL, fn, fal= se, -1); + ((Aggref *)node)->args =3D aggref->args; + + tle_count =3D makeTargetEntry((Expr *) node, + *next_resno, + pstrdup(makeObjectName("__ivm_sum",resname, "_")), + false); + *aggs =3D lappend(*aggs, tle_count); + (*next_resno)++; + } +} + /* * GetIntoRelEFlags --- compute executor flags needed for CREATE TABLE AS * @@ -943,11 +1073,13 @@ CreateIvmTrigger(Oid relOid, Oid viewOid, int16 type= , int16 timing, bool ex_lock static void check_ivm_restriction(Node *node) { - check_ivm_restriction_walker(node, NULL); + check_ivm_restriction_context context =3D {false}; + + check_ivm_restriction_walker(node, &context); } =20 static bool -check_ivm_restriction_walker(Node *node, void *context) +check_ivm_restriction_walker(Node *node, check_ivm_restriction_context *co= ntext) { if (node =3D=3D NULL) return false; @@ -976,6 +1108,10 @@ check_ivm_restriction_walker(Node *node, void *contex= t) ereport(ERROR, (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), errmsg("CTE is not supported on incrementally maintainable materia= lized view"))); + if (qry->groupClause !=3D NIL && !qry->hasAggs) + ereport(ERROR, + (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg("GROUP BY clause without aggregate is not supported on incr= ementally maintainable materialized view"))); if (qry->havingQual !=3D NULL) ereport(ERROR, (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), @@ -1028,6 +1164,8 @@ check_ivm_restriction_walker(Node *node, void *contex= t) } } =20 + context->has_agg |=3D qry->hasAggs; + /* restrictions for rtable */ foreach(lc, qry->rtable) { @@ -1076,7 +1214,7 @@ check_ivm_restriction_walker(Node *node, void *contex= t) =20 } =20 - query_tree_walker(qry, check_ivm_restriction_walker, NULL, QTW_IGNORE_= RANGE_TABLE); + query_tree_walker(qry, check_ivm_restriction_walker, (void *) context,= QTW_IGNORE_RANGE_TABLE); =20 break; } @@ -1087,8 +1225,12 @@ check_ivm_restriction_walker(Node *node, void *conte= xt) ereport(ERROR, (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), errmsg("column name %s is not supported on incrementally maintain= able materialized view", tle->resname))); + if (context->has_agg && !IsA(tle->expr, Aggref) && contain_aggs_of_lev= el((Node *) tle->expr, 0)) + ereport(ERROR, + (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg("expression containing an aggregate in it is not supported = on incrementally maintainable materialized view"))); =20 - expression_tree_walker(node, check_ivm_restriction_walker, NULL); + expression_tree_walker(node, check_ivm_restriction_walker, (void *) co= ntext); break; } case T_JoinExpr: @@ -1100,14 +1242,36 @@ check_ivm_restriction_walker(Node *node, void *cont= ext) (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), errmsg("OUTER JOIN is not supported on incrementally maintainable= materialized view"))); =20 - expression_tree_walker(node, check_ivm_restriction_walker, NULL); + expression_tree_walker(node, check_ivm_restriction_walker, (void *) co= ntext); + break; } - break; case T_Aggref: - ereport(ERROR, - (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), - errmsg("aggregate function is not supported on incrementally maintai= nable materialized view"))); - break; + { + /* Check if this supports IVM */ + Aggref *aggref =3D (Aggref *) node; + const char *aggname =3D format_procedure(aggref->aggfnoid); + + if (aggref->aggfilter !=3D NULL) + ereport(ERROR, + (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg("aggregate function with FILTER clause is not supported on = incrementally maintainable materialized view"))); + + if (aggref->aggdistinct !=3D NULL) + ereport(ERROR, + (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg("aggregate function with DISTINCT arguments is not supporte= d on incrementally maintainable materialized view"))); + + if (aggref->aggorder !=3D NULL) + ereport(ERROR, + (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg("aggregate function with ORDER clause is not supported on i= ncrementally maintainable materialized view"))); + + if (!check_aggregate_supports_ivm(aggref->aggfnoid)) + ereport(ERROR, + (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg("aggregate function %s is not supported on incrementally ma= intainable materialized view", aggname))); + break; + } default: expression_tree_walker(node, check_ivm_restriction_walker, (void *) con= text); break; @@ -1115,6 +1279,46 @@ check_ivm_restriction_walker(Node *node, void *conte= xt) return false; } =20 +/* + * check_aggregate_supports_ivm + * + * Check if the given aggregate function is supporting IVM + */ +static bool +check_aggregate_supports_ivm(Oid aggfnoid) +{ + switch (aggfnoid) + { + /* count */ + case F_COUNT_ANY: + case F_COUNT_: + + /* sum */ + case F_SUM_INT8: + case F_SUM_INT4: + case F_SUM_INT2: + case F_SUM_FLOAT4: + case F_SUM_FLOAT8: + case F_SUM_MONEY: + case F_SUM_INTERVAL: + case F_SUM_NUMERIC: + + /* avg */ + case F_AVG_INT8: + case F_AVG_INT4: + case F_AVG_INT2: + case F_AVG_NUMERIC: + case F_AVG_FLOAT4: + case F_AVG_FLOAT8: + case F_AVG_INTERVAL: + + return true; + + default: + return false; + } +} + /* * CreateIndexOnIMMV * @@ -1172,7 +1376,29 @@ CreateIndexOnIMMV(Query *query, Relation matviewRel) index->concurrent =3D false; index->if_not_exists =3D false; =20 - if (query->distinctClause) + if (query->groupClause) + { + /* create unique constraint on GROUP BY expression columns */ + foreach(lc, query->groupClause) + { + SortGroupClause *scl =3D (SortGroupClause *) lfirst(lc); + TargetEntry *tle =3D get_sortgroupclause_tle(scl, query->targetList); + Form_pg_attribute attr =3D TupleDescAttr(matviewRel->rd_att, tle->resno= - 1); + IndexElem *iparam; + + iparam =3D makeNode(IndexElem); + iparam->name =3D pstrdup(NameStr(attr->attname)); + iparam->expr =3D NULL; + iparam->indexcolname =3D NULL; + iparam->collation =3D NIL; + iparam->opclass =3D NIL; + iparam->opclassopts =3D NIL; + iparam->ordering =3D SORTBY_DEFAULT; + iparam->nulls_ordering =3D SORTBY_NULLS_DEFAULT; + index->indexParams =3D lappend(index->indexParams, iparam); + } + } + else if (query->distinctClause) { /* create unique constraint on all columns */ foreach(lc, query->targetList) @@ -1230,7 +1456,7 @@ CreateIndexOnIMMV(Query *query, Relation matviewRel) (errmsg("could not create an index on materialized view \"%s\" automa= tically", RelationGetRelationName(matviewRel)), errdetail("This target list does not have all the primary key column= s, " - "or this view does not contain DISTINCT clause."), + "or this view does not contain GROUP BY or DISTINCT clause."), errhint("Create an index on the materialized view for efficient incr= emental maintenance."))); return; } diff --git a/src/backend/commands/matview.c b/src/backend/commands/matview.c index dbcbc79fff..3c523991ed 100644 --- a/src/backend/commands/matview.c +++ b/src/backend/commands/matview.c @@ -30,6 +30,7 @@ #include "catalog/pg_opclass.h" #include "catalog/pg_operator.h" #include "commands/cluster.h" +#include "commands/defrem.h" #include "commands/matview.h" #include "commands/tablecmds.h" #include "commands/tablespace.h" @@ -39,6 +40,7 @@ #include "executor/tstoreReceiver.h" #include "miscadmin.h" #include "nodes/makefuncs.h" +#include "optimizer/optimizer.h" #include "parser/analyze.h" #include "parser/parse_clause.h" #include "parser/parse_func.h" @@ -111,6 +113,13 @@ static HTAB *mv_trigger_info =3D NULL; =20 static bool in_delta_calculation =3D false; =20 +/* kind of IVM operation for the view */ +typedef enum +{ + IVM_ADD, + IVM_SUB +} IvmOp; + /* ENR name for materialized view delta */ #define NEW_DELTA_ENRNAME "new_delta" #define OLD_DELTA_ENRNAME "old_delta" @@ -142,7 +151,7 @@ static RangeTblEntry *get_prestate_rte(RangeTblEntry *r= te, MV_TriggerTable *tabl QueryEnvironment *queryEnv, Oid matviewid); static RangeTblEntry *replace_rte_with_delta(RangeTblEntry *rte, MV_Trigge= rTable *table, bool is_new, QueryEnvironment *queryEnv); -static Query *rewrite_query_for_counting(Query *query, ParseState *pstate); +static Query *rewrite_query_for_counting_and_aggregates(Query *query, Pars= eState *pstate); =20 static void calc_delta(MV_TriggerTable *table, int rte_index, Query *query, DestReceiver *dest_old, DestReceiver *dest_new, @@ -153,14 +162,27 @@ static Query *rewrite_query_for_postupdate_state(Quer= y *query, MV_TriggerTable * static void apply_delta(Oid matviewOid, Tuplestorestate *old_tuplestores, = Tuplestorestate *new_tuplestores, TupleDesc tupdesc_old, TupleDesc tupdesc_new, Query *query, bool use_count, char *count_colname); +static void append_set_clause_for_count(const char *resname, StringInfo bu= f_old, + StringInfo buf_new,StringInfo aggs_list); +static void append_set_clause_for_sum(const char *resname, StringInfo buf_= old, + StringInfo buf_new, StringInfo aggs_list); +static void append_set_clause_for_avg(const char *resname, StringInfo buf_= old, + StringInfo buf_new, StringInfo aggs_list, + const char *aggtype); +static char *get_operation_string(IvmOp op, const char *col, const char *a= rg1, const char *arg2, + const char* count_col, const char *castType); +static char *get_null_condition_string(IvmOp op, const char *arg1, const c= har *arg2, + const char* count_col); static void apply_old_delta(const char *matviewname, const char *deltaname= _old, List *keys); static void apply_old_delta_with_count(const char *matviewname, const char= *deltaname_old, - List *keys, const char *count_colname); + List *keys, StringInfo aggs_list, StringInfo aggs_set, + const char *count_colname); static void apply_new_delta(const char *matviewname, const char *deltaname= _new, StringInfo target_list); static void apply_new_delta_with_count(const char *matviewname, const char= * deltaname_new, - List *keys, StringInfo target_list, const char* count_colname); + List *keys, StringInfo target_list, StringInfo aggs_set, + const char* count_colname); static char *get_matching_condition_string(List *keys); static void generate_equal(StringInfo querybuf, Oid opttype, const char *leftop, const char *rightop); @@ -1453,11 +1475,44 @@ IVM_immediate_maintenance(PG_FUNCTION_ARGS) * When a base table is truncated, the view content will be empty if the * view definition query does not contain an aggregate without a GROUP cl= ause. * Therefore, such views can be truncated. + * + * Aggregate views without a GROUP clause always have one row. Therefore, + * if a base table is truncated, the view will not be empty and will cont= ain + * a row with NULL value (or 0 for count()). So, in this case, we refresh= the + * view instead of truncating it. */ if (TRIGGER_FIRED_BY_TRUNCATE(trigdata->tg_event)) { - ExecuteTruncateGuts(list_make1(matviewRel), list_make1_oid(matviewOid), - NIL, DROP_RESTRICT, false, false); + if (!(query->hasAggs && query->groupClause =3D=3D NIL)) + ExecuteTruncateGuts(list_make1(matviewRel), list_make1_oid(matviewOid), + NIL, DROP_RESTRICT, false, false); + else + { + Oid OIDNewHeap; + DestReceiver *dest; + uint64 processed =3D 0; + Query *dataQuery =3D rewriteQueryForIMMV(query, NIL); + char relpersistence =3D matviewRel->rd_rel->relpersistence; + + /* + * Create the transient table that will receive the regenerated data. L= ock + * it against access by any other process until commit (by which time it + * will be gone). + */ + OIDNewHeap =3D make_new_heap(matviewOid, matviewRel->rd_rel->reltablesp= ace, + matviewRel->rd_rel->relam, + relpersistence, ExclusiveLock); + LockRelationOid(OIDNewHeap, AccessExclusiveLock); + dest =3D CreateTransientRelDestReceiver(OIDNewHeap); + + /* Generate the data */ + processed =3D refresh_matview_datafill(dest, dataQuery, NULL, NULL, ""); + refresh_by_heap_swap(matviewOid, OIDNewHeap, relpersistence); + + /* Inform cumulative stats system about our activity */ + pgstat_count_truncate(matviewRel); + pgstat_count_heap_insert(matviewRel, processed); + } =20 /* Clean up hash entry and delete tuplestores */ clean_up_IVM_hash_entry(entry, false); @@ -1497,8 +1552,8 @@ IVM_immediate_maintenance(PG_FUNCTION_ARGS) /* Set all tables in the query to pre-update state */ rewritten =3D rewrite_query_for_preupdate_state(rewritten, entry->tables, pstate, matviewOid); - /* Rewrite for counting duplicated tuples */ - rewritten =3D rewrite_query_for_counting(rewritten, pstate); + /* Rewrite for counting duplicated tuples and aggregates functions*/ + rewritten =3D rewrite_query_for_counting_and_aggregates(rewritten, pstate= ); =20 /* Create tuplestores to store view deltas */ if (entry->has_old) @@ -1549,7 +1604,7 @@ IVM_immediate_maintenance(PG_FUNCTION_ARGS) =20 count_colname =3D pstrdup("__ivm_count__"); =20 - if (query->distinctClause) + if (query->hasAggs || query->distinctClause) use_count =3D true; =20 /* calculate delta tables */ @@ -1945,17 +2000,34 @@ replace_rte_with_delta(RangeTblEntry *rte, MV_Trigg= erTable *table, bool is_new, } =20 /* - * rewrite_query_for_counting + * rewrite_query_for_counting_and_aggregates * - * Rewrite query for counting duplicated tuples. + * Rewrite query for counting duplicated tuples and aggregate functions. */ static Query * -rewrite_query_for_counting(Query *query, ParseState *pstate) +rewrite_query_for_counting_and_aggregates(Query *query, ParseState *pstate) { TargetEntry *tle_count; FuncCall *fn; Node *node; =20 + /* For aggregate views */ + if (query->hasAggs) + { + ListCell *lc; + List *aggs =3D NIL; + AttrNumber next_resno =3D list_length(query->targetList) + 1; + + foreach(lc, query->targetList) + { + TargetEntry *tle =3D (TargetEntry *) lfirst(lc); + + if (IsA(tle->expr, Aggref)) + makeIvmAggColumn(pstate, (Aggref *)tle->expr, tle->resname, &next_resn= o, &aggs); + } + query->targetList =3D list_concat(query->targetList, aggs); + } + /* Add count(*) for counting distinct tuples in views */ fn =3D makeFuncCall(SystemFuncName("count"), NIL, COERCE_EXPLICIT_CALL, -= 1); fn->agg_star =3D true; @@ -2028,6 +2100,8 @@ rewrite_query_for_postupdate_state(Query *query, MV_T= riggerTable *table, int rte return query; } =20 +#define IVM_colname(type, col) makeObjectName("__ivm_" type, col, "_") + /* * apply_delta * @@ -2041,6 +2115,9 @@ apply_delta(Oid matviewOid, Tuplestorestate *old_tupl= estores, Tuplestorestate *n { StringInfoData querybuf; StringInfoData target_list_buf; + StringInfo aggs_list_buf =3D NULL; + StringInfo aggs_set_old =3D NULL; + StringInfo aggs_set_new =3D NULL; Relation matviewRel; char *matviewname; ListCell *lc; @@ -2063,6 +2140,15 @@ apply_delta(Oid matviewOid, Tuplestorestate *old_tup= lestores, Tuplestorestate *n initStringInfo(&querybuf); initStringInfo(&target_list_buf); =20 + if (query->hasAggs) + { + if (old_tuplestores && tuplestore_tuple_count(old_tuplestores) > 0) + aggs_set_old =3D makeStringInfo(); + if (new_tuplestores && tuplestore_tuple_count(new_tuplestores) > 0) + aggs_set_new =3D makeStringInfo(); + aggs_list_buf =3D makeStringInfo(); + } + /* build string of target list */ for (i =3D 0; i < matviewRel->rd_att->natts; i++) { @@ -2079,13 +2165,61 @@ apply_delta(Oid matviewOid, Tuplestorestate *old_tu= plestores, Tuplestorestate *n { TargetEntry *tle =3D (TargetEntry *) lfirst(lc); Form_pg_attribute attr =3D TupleDescAttr(matviewRel->rd_att, i); + char *resname =3D NameStr(attr->attname); =20 i++; =20 if (tle->resjunk) continue; =20 - keys =3D lappend(keys, attr); + /* + * For views without aggregates, all attributes are used as keys to iden= tify a + * tuple in a view. + */ + if (!query->hasAggs) + keys =3D lappend(keys, attr); + + /* For views with aggregates, we need to build SET clause for updating a= ggregate + * values. */ + if (query->hasAggs && IsA(tle->expr, Aggref)) + { + Aggref *aggref =3D (Aggref *) tle->expr; + const char *aggname =3D get_func_name(aggref->aggfnoid); + + /* + * We can use function names here because it is already checked if these + * can be used in IMMV by its OID at the definition time. + */ + + /* count */ + if (!strcmp(aggname, "count")) + append_set_clause_for_count(resname, aggs_set_old, aggs_set_new, aggs_= list_buf); + + /* sum */ + else if (!strcmp(aggname, "sum")) + append_set_clause_for_sum(resname, aggs_set_old, aggs_set_new, aggs_li= st_buf); + + /* avg */ + else if (!strcmp(aggname, "avg")) + append_set_clause_for_avg(resname, aggs_set_old, aggs_set_new, aggs_li= st_buf, + format_type_be(aggref->aggtype)); + + else + elog(ERROR, "unsupported aggregate function: %s", aggname); + } + } + + /* If we have GROUP BY clause, we use its entries as keys. */ + if (query->hasAggs && query->groupClause) + { + foreach (lc, query->groupClause) + { + SortGroupClause *sgcl =3D (SortGroupClause *) lfirst(lc); + TargetEntry *tle =3D get_sortgroupclause_tle(sgcl, query->targetList); + Form_pg_attribute attr =3D TupleDescAttr(matviewRel->rd_att, tle->resno= - 1); + + keys =3D lappend(keys, attr); + } } =20 /* Start maintaining the materialized view. */ @@ -2116,7 +2250,8 @@ apply_delta(Oid matviewOid, Tuplestorestate *old_tupl= estores, Tuplestorestate *n if (use_count) /* apply old delta and get rows to be recalculated */ apply_old_delta_with_count(matviewname, OLD_DELTA_ENRNAME, - keys, count_colname); + keys, aggs_list_buf, aggs_set_old, + count_colname); else apply_old_delta(matviewname, OLD_DELTA_ENRNAME, keys); =20 @@ -2142,7 +2277,7 @@ apply_delta(Oid matviewOid, Tuplestorestate *old_tupl= estores, Tuplestorestate *n /* apply new delta */ if (use_count) apply_new_delta_with_count(matviewname, NEW_DELTA_ENRNAME, - keys, &target_list_buf, count_colname); + keys, aggs_set_new, &target_list_buf, count_colname); else apply_new_delta(matviewname, NEW_DELTA_ENRNAME, &target_list_buf); } @@ -2157,6 +2292,250 @@ apply_delta(Oid matviewOid, Tuplestorestate *old_tu= plestores, Tuplestorestate *n elog(ERROR, "SPI_finish failed"); } =20 +/* + * append_set_clause_for_count + * + * Append SET clause string for count aggregation to given buffers. + * Also, append resnames required for calculating the aggregate value. + */ +static void +append_set_clause_for_count(const char *resname, StringInfo buf_old, + StringInfo buf_new,StringInfo aggs_list) +{ + /* For tuple deletion */ + if (buf_old) + { + /* resname =3D mv.resname - t.resname */ + appendStringInfo(buf_old, + ", %s =3D %s", + quote_qualified_identifier(NULL, resname), + get_operation_string(IVM_SUB, resname, "mv", "t", NULL, NULL)); + } + /* For tuple insertion */ + if (buf_new) + { + /* resname =3D mv.resname + diff.resname */ + appendStringInfo(buf_new, + ", %s =3D %s", + quote_qualified_identifier(NULL, resname), + get_operation_string(IVM_ADD, resname, "mv", "diff", NULL, NULL)); + } + + appendStringInfo(aggs_list, ", %s", + quote_qualified_identifier("diff", resname) + ); +} + +/* + * append_set_clause_for_sum + * + * Append SET clause string for sum aggregation to given buffers. + * Also, append resnames required for calculating the aggregate value. + */ +static void +append_set_clause_for_sum(const char *resname, StringInfo buf_old, + StringInfo buf_new, StringInfo aggs_list) +{ + char *count_col =3D IVM_colname("count", resname); + + /* For tuple deletion */ + if (buf_old) + { + /* sum =3D mv.sum - t.sum */ + appendStringInfo(buf_old, + ", %s =3D %s", + quote_qualified_identifier(NULL, resname), + get_operation_string(IVM_SUB, resname, "mv", "t", count_col, NULL) + ); + /* count =3D mv.count - t.count */ + appendStringInfo(buf_old, + ", %s =3D %s", + quote_qualified_identifier(NULL, count_col), + get_operation_string(IVM_SUB, count_col, "mv", "t", NULL, NULL) + ); + } + /* For tuple insertion */ + if (buf_new) + { + /* sum =3D mv.sum + diff.sum */ + appendStringInfo(buf_new, + ", %s =3D %s", + quote_qualified_identifier(NULL, resname), + get_operation_string(IVM_ADD, resname, "mv", "diff", count_col, NULL) + ); + /* count =3D mv.count + diff.count */ + appendStringInfo(buf_new, + ", %s =3D %s", + quote_qualified_identifier(NULL, count_col), + get_operation_string(IVM_ADD, count_col, "mv", "diff", NULL, NULL) + ); + } + + appendStringInfo(aggs_list, ", %s, %s", + quote_qualified_identifier("diff", resname), + quote_qualified_identifier("diff", IVM_colname("count", resname)) + ); +} + +/* + * append_set_clause_for_avg + * + * Append SET clause string for avg aggregation to given buffers. + * Also, append resnames required for calculating the aggregate value. + */ +static void +append_set_clause_for_avg(const char *resname, StringInfo buf_old, + StringInfo buf_new, StringInfo aggs_list, + const char *aggtype) +{ + char *sum_col =3D IVM_colname("sum", resname); + char *count_col =3D IVM_colname("count", resname); + + /* For tuple deletion */ + if (buf_old) + { + /* avg =3D (mv.sum - t.sum)::aggtype / (mv.count - t.count) */ + appendStringInfo(buf_old, + ", %s =3D %s OPERATOR(pg_catalog./) %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 =3D mv.sum - t.sum */ + appendStringInfo(buf_old, + ", %s =3D %s", + quote_qualified_identifier(NULL, sum_col), + get_operation_string(IVM_SUB, sum_col, "mv", "t", count_col, NULL) + ); + /* count =3D mv.count - t.count */ + appendStringInfo(buf_old, + ", %s =3D %s", + quote_qualified_identifier(NULL, count_col), + get_operation_string(IVM_SUB, count_col, "mv", "t", NULL, NULL) + ); + + } + /* For tuple insertion */ + if (buf_new) + { + /* avg =3D (mv.sum + diff.sum)::aggtype / (mv.count + diff.count) */ + appendStringInfo(buf_new, + ", %s =3D %s OPERATOR(pg_catalog./) %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 =3D mv.sum + diff.sum */ + appendStringInfo(buf_new, + ", %s =3D %s", + quote_qualified_identifier(NULL, sum_col), + get_operation_string(IVM_ADD, sum_col, "mv", "diff", count_col, NULL) + ); + /* count =3D mv.count + diff.count */ + appendStringInfo(buf_new, + ", %s =3D %s", + quote_qualified_identifier(NULL, count_col), + get_operation_string(IVM_ADD, count_col, "mv", "diff", NULL, NULL) + ); + } + + appendStringInfo(aggs_list, ", %s, %s, %s", + quote_qualified_identifier("diff", resname), + quote_qualified_identifier("diff", IVM_colname("sum", resname)), + quote_qualified_identifier("diff", IVM_colname("count", resname)) + ); +} + +/* + * get_operation_string + * + * Build a string to calculate the new aggregate values. + */ +static char * +get_operation_string(IvmOp op, const char *col, const char *arg1, const ch= ar *arg2, + const char* count_col, const char *castType) +{ + StringInfoData buf; + StringInfoData castString; + char *col1 =3D quote_qualified_identifier(arg1, col); + char *col2 =3D quote_qualified_identifier(arg2, col); + char op_char =3D (op =3D=3D IVM_SUB ? '-' : '+'); + + initStringInfo(&buf); + initStringInfo(&castString); + + if (castType) + appendStringInfo(&castString, "::%s", castType); + + if (!count_col) + { + /* + * If the attributes don't have count columns then calc the result + * by using the operator simply. + */ + appendStringInfo(&buf, "(%s OPERATOR(pg_catalog.%c) %s)%s", + col1, op_char, col2, castString.data); + } + else + { + /* + * If the attributes have count columns then consider the condition + * where the result becomes NULL. + */ + char *null_cond =3D get_null_condition_string(op, arg1, arg2, count_col); + + appendStringInfo(&buf, + "(CASE WHEN %s THEN NULL " + "WHEN %s IS NULL THEN %s " + "WHEN %s IS NULL THEN %s " + "ELSE (%s OPERATOR(pg_catalog.%c) %s)%s END)", + null_cond, + col1, col2, + col2, col1, + col1, op_char, col2, castString.data + ); + } + + return buf.data; +} + +/* + * get_null_condition_string + * + * Build a predicate string for CASE clause to check if an aggregate value + * will became NULL after the given operation is applied. + */ +static char * +get_null_condition_string(IvmOp op, const char *arg1, const char *arg2, + const char* count_col) +{ + StringInfoData null_cond; + initStringInfo(&null_cond); + + switch (op) + { + case IVM_ADD: + appendStringInfo(&null_cond, + "%s OPERATOR(pg_catalog.=3D) 0 AND %s OPERATOR(pg_catalog.=3D) 0", + quote_qualified_identifier(arg1, count_col), + quote_qualified_identifier(arg2, count_col) + ); + break; + case IVM_SUB: + appendStringInfo(&null_cond, + "%s OPERATOR(pg_catalog.=3D) %s", + quote_qualified_identifier(arg1, count_col), + quote_qualified_identifier(arg2, count_col) + ); + break; + default: + elog(ERROR,"unknown operation"); + } + + return null_cond.data; +} + + /* * apply_old_delta_with_count * @@ -2164,13 +2543,20 @@ apply_delta(Oid matviewOid, Tuplestorestate *old_tu= plestores, Tuplestorestate *n * which contains tuples to be deleted from to a materialized view given by * matviewname. This is used when counting is required, that is, the view * has aggregate or distinct. + * + * If the view desn't have aggregates or has GROUP BY, this requires a keys + * list to identify a tuple in the view. If the view has aggregates, this + * requires strings representing resnames of aggregates and SET clause for + * updating aggregate values. */ static void apply_old_delta_with_count(const char *matviewname, const char *deltaname_= old, - List *keys, const char *count_colname) + List *keys, StringInfo aggs_list, StringInfo aggs_set, + const char *count_colname) { StringInfoData querybuf; char *match_cond; + bool agg_without_groupby =3D (list_length(keys) =3D=3D 0); =20 /* build WHERE condition for searching tuples to be deleted */ match_cond =3D get_matching_condition_string(keys); @@ -2180,22 +2566,26 @@ apply_old_delta_with_count(const char *matviewname,= const char *deltaname_old, appendStringInfo(&querybuf, "WITH t AS (" /* collecting tid of target tuples in the view */ "SELECT diff.%s, " /* count column */ - "(diff.%s OPERATOR(pg_catalog.=3D) mv.%s) AS for_dlt, " + "(diff.%s OPERATOR(pg_catalog.=3D) mv.%s AND %s) AS for_dlt, " "mv.ctid " + "%s " /* aggregate columns */ "FROM %s AS mv, %s AS diff " "WHERE %s" /* tuple matching condition */ "), updt AS (" /* update a tuple if this is not to be deleted */ "UPDATE %s AS mv SET %s =3D mv.%s OPERATOR(pg_catalog.-) t.%s " + "%s" /* SET clauses for aggregates */ "FROM t WHERE mv.ctid OPERATOR(pg_catalog.=3D) t.ctid AND NOT for_dl= t " ")" /* delete a tuple if this is to be deleted */ "DELETE FROM %s AS mv USING t " "WHERE mv.ctid OPERATOR(pg_catalog.=3D) t.ctid AND for_dlt", count_colname, - count_colname, count_colname, + count_colname, count_colname, (agg_without_groupby ? "false" : "true"= ), + (aggs_list !=3D NULL ? aggs_list->data : ""), matviewname, deltaname_old, match_cond, matviewname, count_colname, count_colname, count_colname, + (aggs_set !=3D NULL ? aggs_set->data : ""), matviewname); =20 if (SPI_exec(querybuf.data, 0) !=3D SPI_OK_DELETE) @@ -2259,10 +2649,15 @@ apply_old_delta(const char *matviewname, const char= *deltaname_old, * matviewname. This is used when counting is required, that is, the view * has aggregate or distinct. Also, when a table in EXISTS sub queries * is modified. + * + * If the view desn't have aggregates or has GROUP BY, this requires a keys + * list to identify a tuple in the view. If the view has aggregates, this + * requires strings representing SET clause for updating aggregate values. */ static void apply_new_delta_with_count(const char *matviewname, const char* deltaname_= new, - List *keys, StringInfo target_list, const char* count_colname) + List *keys, StringInfo aggs_set, StringInfo target_list, + const char* count_colname) { StringInfoData querybuf; StringInfoData returning_keys; @@ -2293,6 +2688,7 @@ apply_new_delta_with_count(const char *matviewname, c= onst char* deltaname_new, appendStringInfo(&querybuf, "WITH updt AS (" /* update a tuple if this exists in the view */ "UPDATE %s AS mv SET %s =3D mv.%s OPERATOR(pg_catalog.+) diff.%s " + "%s " /* SET clauses for aggregates */ "FROM %s AS diff " "WHERE %s " /* tuple matching condition */ "RETURNING %s" /* returning keys of updated tuples */ @@ -2300,6 +2696,7 @@ apply_new_delta_with_count(const char *matviewname, c= onst char* deltaname_new, "SELECT %s FROM %s AS diff " "WHERE NOT EXISTS (SELECT 1 FROM updt AS mv WHERE %s);", matviewname, count_colname, count_colname, count_colname, + (aggs_set !=3D NULL ? aggs_set->data : ""), deltaname_new, match_cond, returning_keys.data, diff --git a/src/include/commands/createas.h b/src/include/commands/createa= s.h index 6b47e66bfd..af3a5b4b27 100644 --- a/src/include/commands/createas.h +++ b/src/include/commands/createas.h @@ -30,6 +30,7 @@ extern void CreateIvmTriggersOnBaseTables(Query *qry, Oid= matviewOid); extern void CreateIndexOnIMMV(Query *query, Relation matviewRel); =20 extern Query *rewriteQueryForIMMV(Query *query, List *colNames); +extern void makeIvmAggColumn(ParseState *pstate, Aggref *aggref, char *res= name, AttrNumber *next_resno, List **aggs); =20 extern int GetIntoRelEFlags(IntoClause *intoClause); =20 --=20 2.25.1 --Multipart=_Mon__4_Mar_2024_11_58_46_+0900_UaponF/qQhQrVCFt Content-Type: text/x-diff; name="v30-0009-Add-support-for-min-max-aggregates-for-IVM.patch" Content-Disposition: attachment; filename="v30-0009-Add-support-for-min-max-aggregates-for-IVM.patch" Content-Transfer-Encoding: 7bit ^ permalink raw reply [nested|flat] 11+ messages in thread
* Fix premature xmin advancement during fast forward decoding @ 2025-04-22 07:06 Zhijie Hou (Fujitsu) <[email protected]> 2025-04-23 06:31 ` Re: Fix premature xmin advancement during fast forward decoding shveta malik <[email protected]> 2025-04-25 02:54 ` RE: Fix premature xmin advancement during fast forward decoding Zhijie Hou (Fujitsu) <[email protected]> 2025-04-26 12:01 ` Re: Fix premature xmin advancement during fast forward decoding Amit Kapila <[email protected]> 0 siblings, 3 replies; 11+ messages in thread From: Zhijie Hou (Fujitsu) @ 2025-04-22 07:06 UTC (permalink / raw) To: PostgreSQL Hackers <[email protected]>; +Cc: Amit Kapila <[email protected]>; Masahiko Sawada <[email protected]>; shveta malik <[email protected]> Hi, When analyzing some issues in another thread[1], I found a bug that fast forward decoding could lead to premature advancement of catalog_xmin, resulting in required catalog data being removed by vacuum. The issue arises because we do not build a base snapshot when decoding changes during fast forward mode, preventing reference to the minimum transaction ID that remains visible in the snapshot when determining the candidate for catalog_xmin. As a result, catalog_xmin was directly advanced to the oldest running transaction ID found in the latest running_xacts record. In code-level, SnapBuildProcessChange -> ReorderBufferSetBaseSnapshot could not be reached during fast forward decoding, resulting in rb->txns_by_base_snapshot_lsn being NULL. When advancing catalog_xmin, the system attempts to refer to rb->txns_by_base_snapshot_lsn via ReorderBufferGetOldestXmin(). However, since rb->txns_by_base_snapshot_lsn is NULL, it defaults to directly using running->oldestRunningXid as the candidate for catalog_xmin. For reference, see the details in SnapBuildProcessRunningXacts. See the attachment for a test(0002) to prove that the catalog data that are still required would be removed after premature catalog_xmin advancement during fast forward decoding. To fix this, I think we can allow the base snapshot to be built during fast forward decoding, as implemented in the patch 0001 (We already built base snapshot in fast-forward mode for logical message in logicalmsg_decode()). I also explored the possibility of further optimizing the fix to reduce the overhead associated with building a snapshot in fast-forward mode. E.g., to maintain a single base_snapshot_xmin rather than generating a full snapshot for each transaction, and use this base_snapshot_xmin when determining the candidate catalog_xmin. However, I am not feeling confident about maintaining some fast-forward dedicated fields in common structures and perhaps employing different handling for catalog_xmin. Moreover, I conducted a basic test[2] to test the patch's impact, noting that advancing the slot incurs roughly a 4% increase in processing time after applying the patch, which appears to be acceptable. Additionally, the cost associated with building the snapshot via SnapBuildBuildSnapshot() did not show up in the profile. Therefore, I think it's reasonable to refrain from further optimization at this stage. [1] https://www.postgresql.org/message-id/OS3PR01MB5718ED3F0123737C7380BBAD94BB2%40OS3PR01MB5718.jpnprd0... [2] Create a table test(a int); Create a slot A. pgbench postgres -T 60 -j 100 -c 100 -f simple-insert simple-insert: BEGIN; INSERT INTO test VALUES(1); COMMIT; use pg_replication_slot_advance to advance the slot A to the latest position and count the time. HEAD: Time: 4189.257 ms PATCH: Time: 4371.358 ms Best Regards, Hou zj Attachments: [application/octet-stream] v1-0002-Add-a-test.patch (3.3K, ../../OS0PR01MB57163087F86621D44D9A72BF94BB2@OS0PR01MB5716.jpnprd01.prod.outlook.com/2-v1-0002-Add-a-test.patch) download | inline diff: From f55acc6c9838dbb04f50a7677517214ee43d5be8 Mon Sep 17 00:00:00 2001 From: Zhijie Hou <[email protected]> Date: Tue, 22 Apr 2025 11:03:40 +0800 Subject: [PATCH v1 2/2] Add a test --- .../test_decoding/expected/oldest_xmin.out | 41 +++++++++++++++++++ contrib/test_decoding/specs/oldest_xmin.spec | 5 +++ 2 files changed, 46 insertions(+) diff --git a/contrib/test_decoding/expected/oldest_xmin.out b/contrib/test_decoding/expected/oldest_xmin.out index dd6053f9c1f..57268b38d33 100644 --- a/contrib/test_decoding/expected/oldest_xmin.out +++ b/contrib/test_decoding/expected/oldest_xmin.out @@ -38,3 +38,44 @@ COMMIT stop (1 row) + +starting permutation: s0_begin s0_getxid s1_begin s1_insert s0_alter s0_commit s0_checkpoint s0_advance_slot s0_advance_slot s1_commit s0_vacuum s0_get_changes +step s0_begin: BEGIN; +step s0_getxid: SELECT pg_current_xact_id() IS NULL; +?column? +-------- +f +(1 row) + +step s1_begin: BEGIN; +step s1_insert: INSERT INTO harvest VALUES ((1, 2, 3)); +step s0_alter: ALTER TYPE basket DROP ATTRIBUTE mangos; +step s0_commit: COMMIT; +step s0_checkpoint: CHECKPOINT; +step s0_advance_slot: SELECT slot_name FROM pg_replication_slot_advance('isolation_slot', pg_current_wal_lsn()); +slot_name +-------------- +isolation_slot +(1 row) + +step s0_advance_slot: SELECT slot_name FROM pg_replication_slot_advance('isolation_slot', pg_current_wal_lsn()); +slot_name +-------------- +isolation_slot +(1 row) + +step s1_commit: COMMIT; +step s0_vacuum: VACUUM pg_attribute; +step s0_get_changes: SELECT data FROM pg_logical_slot_get_changes('isolation_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1'); +data +------------------------------------------------------ +BEGIN +table public.harvest: INSERT: fruits[basket]:'(1,2,3)' +COMMIT +(3 rows) + +?column? +-------- +stop +(1 row) + diff --git a/contrib/test_decoding/specs/oldest_xmin.spec b/contrib/test_decoding/specs/oldest_xmin.spec index 88bd30f5ff7..7f2fe3d7ed7 100644 --- a/contrib/test_decoding/specs/oldest_xmin.spec +++ b/contrib/test_decoding/specs/oldest_xmin.spec @@ -25,6 +25,7 @@ step "s0_commit" { COMMIT; } step "s0_checkpoint" { CHECKPOINT; } step "s0_vacuum" { VACUUM pg_attribute; } step "s0_get_changes" { SELECT data FROM pg_logical_slot_get_changes('isolation_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1'); } +step "s0_advance_slot" { SELECT slot_name FROM pg_replication_slot_advance('isolation_slot', pg_current_wal_lsn()); } session "s1" setup { SET synchronous_commit=on; } @@ -40,3 +41,7 @@ step "s1_commit" { COMMIT; } # will be removed (xmax set) before T1 commits. That is, interlocking doesn't # forbid modifying catalog after someone read it (and didn't commit yet). permutation "s0_begin" "s0_getxid" "s1_begin" "s1_insert" "s0_alter" "s0_commit" "s0_checkpoint" "s0_get_changes" "s0_get_changes" "s1_commit" "s0_vacuum" "s0_get_changes" + +# Perform the same testing process as described above, but use advance_slot to +# forces xmin advancement during fast forward decoding. +permutation "s0_begin" "s0_getxid" "s1_begin" "s1_insert" "s0_alter" "s0_commit" "s0_checkpoint" "s0_advance_slot" "s0_advance_slot" "s1_commit" "s0_vacuum" "s0_get_changes" -- 2.30.0.windows.2 [application/octet-stream] v1-0001-Fix-premature-xmin-advancement-during-fast-forwar.patch (4.2K, ../../OS0PR01MB57163087F86621D44D9A72BF94BB2@OS0PR01MB5716.jpnprd01.prod.outlook.com/3-v1-0001-Fix-premature-xmin-advancement-during-fast-forwar.patch) download | inline diff: From 5327186830024769c2d893fba5859ebbdc04af29 Mon Sep 17 00:00:00 2001 From: Zhijie Hou <[email protected]> Date: Tue, 22 Apr 2025 10:58:35 +0800 Subject: [PATCH v1 1/2] Fix premature xmin advancement during fast forward decoding Fast forward decoding could lead to premature advancement of catalog_xmin, resulting in required catalog data being removed by vacuum. The issue arise because we do not build a base snapshot when decoding changes during fast forward mode, preventing reference to the minimum transaction ID that remains visible in the snapshot when determining the candidate for catalog_xmin. As a result, catalog_xmin was directly advanced to the oldest running transaction ID found in the latest running_xacts record. To resolve this, the commit allows the base snapshot to be built during fast forward decoding. --- src/backend/replication/logical/decode.c | 37 ++++++++++++++++-------- 1 file changed, 25 insertions(+), 12 deletions(-) diff --git a/src/backend/replication/logical/decode.c b/src/backend/replication/logical/decode.c index 78f9a0a11c4..8dc467a8bb3 100644 --- a/src/backend/replication/logical/decode.c +++ b/src/backend/replication/logical/decode.c @@ -412,19 +412,24 @@ heap2_decode(LogicalDecodingContext *ctx, XLogRecordBuffer *buf) /* * If we don't have snapshot or we are just fast-forwarding, there is no - * point in decoding changes. + * point in decoding data changes. However, it's crucial to build the base + * snapshot during fast-forward mode (as done in SnapBuildProcessChange()) + * because the snapshot's minimum visible transaction ID (xmin) must be + * referenced when determining the candidate catalog_xmin for the + * replication slot. */ - if (SnapBuildCurrentState(builder) < SNAPBUILD_FULL_SNAPSHOT || - ctx->fast_forward) + if (SnapBuildCurrentState(builder) < SNAPBUILD_FULL_SNAPSHOT) return; switch (info) { case XLOG_HEAP2_MULTI_INSERT: - if (SnapBuildProcessChange(builder, xid, buf->origptr)) + if (SnapBuildProcessChange(builder, xid, buf->origptr) && + !ctx->fast_forward) DecodeMultiInsert(ctx, buf); break; case XLOG_HEAP2_NEW_CID: + if (!ctx->fast_forward) { xl_heap_new_cid *xlrec; @@ -471,16 +476,20 @@ heap_decode(LogicalDecodingContext *ctx, XLogRecordBuffer *buf) /* * If we don't have snapshot or we are just fast-forwarding, there is no - * point in decoding data changes. + * point in decoding data changes. However, it's crucial to build the base + * snapshot during fast-forward mode (as done in SnapBuildProcessChange()) + * because the snapshot's minimum visible transaction ID (xmin) must be + * referenced when determining the candidate catalog_xmin for the + * replication slot. */ - if (SnapBuildCurrentState(builder) < SNAPBUILD_FULL_SNAPSHOT || - ctx->fast_forward) + if (SnapBuildCurrentState(builder) < SNAPBUILD_FULL_SNAPSHOT) return; switch (info) { case XLOG_HEAP_INSERT: - if (SnapBuildProcessChange(builder, xid, buf->origptr)) + if (SnapBuildProcessChange(builder, xid, buf->origptr) && + !ctx->fast_forward) DecodeInsert(ctx, buf); break; @@ -491,17 +500,20 @@ heap_decode(LogicalDecodingContext *ctx, XLogRecordBuffer *buf) */ case XLOG_HEAP_HOT_UPDATE: case XLOG_HEAP_UPDATE: - if (SnapBuildProcessChange(builder, xid, buf->origptr)) + if (SnapBuildProcessChange(builder, xid, buf->origptr) && + !ctx->fast_forward) DecodeUpdate(ctx, buf); break; case XLOG_HEAP_DELETE: - if (SnapBuildProcessChange(builder, xid, buf->origptr)) + if (SnapBuildProcessChange(builder, xid, buf->origptr) && + !ctx->fast_forward) DecodeDelete(ctx, buf); break; case XLOG_HEAP_TRUNCATE: - if (SnapBuildProcessChange(builder, xid, buf->origptr)) + if (SnapBuildProcessChange(builder, xid, buf->origptr) && + !ctx->fast_forward) DecodeTruncate(ctx, buf); break; @@ -525,7 +537,8 @@ heap_decode(LogicalDecodingContext *ctx, XLogRecordBuffer *buf) break; case XLOG_HEAP_CONFIRM: - if (SnapBuildProcessChange(builder, xid, buf->origptr)) + if (SnapBuildProcessChange(builder, xid, buf->origptr) && + !ctx->fast_forward) DecodeSpecConfirm(ctx, buf); break; -- 2.30.0.windows.2 ^ permalink raw reply [nested|flat] 11+ messages in thread
* Re: Fix premature xmin advancement during fast forward decoding 2025-04-22 07:06 Fix premature xmin advancement during fast forward decoding Zhijie Hou (Fujitsu) <[email protected]> @ 2025-04-23 06:31 ` shveta malik <[email protected]> 2025-04-23 08:20 ` RE: Fix premature xmin advancement during fast forward decoding Zhijie Hou (Fujitsu) <[email protected]> 2 siblings, 1 reply; 11+ messages in thread From: shveta malik @ 2025-04-23 06:31 UTC (permalink / raw) To: Zhijie Hou (Fujitsu) <[email protected]>; +Cc: PostgreSQL Hackers <[email protected]>; Amit Kapila <[email protected]>; Masahiko Sawada <[email protected]>; shveta malik <[email protected]> On Tue, Apr 22, 2025 at 12:36 PM Zhijie Hou (Fujitsu) <[email protected]> wrote: > > Hi, > > To fix this, I think we can allow the base snapshot to be built during fast > forward decoding, as implemented in the patch 0001 (We already built base > snapshot in fast-forward mode for logical message in logicalmsg_decode()). The idea and code looks okay to me and the performance impact is also not that huge. IIUC, fast_forward decoding mode is used only in two cases. 1) pg_replication_slot_advance and 2) in upgrade flow to check if there are any pending WAL changes which are not yet replicated. See 'binary_upgrade_logical_slot_has_caught_up'-->'LogicalReplicationSlotHasPendingWal'. It seems like this change will not have any negative impact in the upgrade flow as well (in terms of performance and anything else). Thoughts? > > Moreover, I conducted a basic test[2] to test the patch's impact, noting that > advancing the slot incurs roughly a 4% increase in processing time after > applying the patch, which appears to be acceptable. Additionally, the cost > associated with building the snapshot via SnapBuildBuildSnapshot() did not show > up in the profile. Therefore, I think it's reasonable to refrain from > further optimization at this stage. I agree on this. thanks Shveta ^ permalink raw reply [nested|flat] 11+ messages in thread
* RE: Fix premature xmin advancement during fast forward decoding 2025-04-22 07:06 Fix premature xmin advancement during fast forward decoding Zhijie Hou (Fujitsu) <[email protected]> 2025-04-23 06:31 ` Re: Fix premature xmin advancement during fast forward decoding shveta malik <[email protected]> @ 2025-04-23 08:20 ` Zhijie Hou (Fujitsu) <[email protected]> 0 siblings, 0 replies; 11+ messages in thread From: Zhijie Hou (Fujitsu) @ 2025-04-23 08:20 UTC (permalink / raw) To: shveta malik <[email protected]>; +Cc: PostgreSQL Hackers <[email protected]>; Amit Kapila <[email protected]>; Masahiko Sawada <[email protected]> On Wed, Apr 23, 2025 at 2:31 PM shveta malik wrote: > > On Tue, Apr 22, 2025 at 12:36 PM Zhijie Hou (Fujitsu) > <[email protected]> wrote: > > > > Hi, > > > > To fix this, I think we can allow the base snapshot to be built during fast > > forward decoding, as implemented in the patch 0001 (We already built base > > snapshot in fast-forward mode for logical message in logicalmsg_decode()). > > The idea and code looks okay to me and the performance impact is also > not that huge. Thanks for reviewing! > > IIUC, fast_forward decoding mode is used only in two cases. 1) > pg_replication_slot_advance and 2) in upgrade flow to check if there > are any pending WAL changes which are not yet replicated. See > 'binary_upgrade_logical_slot_has_caught_up'-->'LogicalReplicationSlotHasP > endingWal'. > It seems like this change will not have any negative impact in the > upgrade flow as well (in terms of performance and anything else). > Thoughts? I also think so. The upgrade uses fast forward decoding to check if there are any pending WALs not sent yet and report an ERROR if so. This indicates that we do not expect to decode real changes for the upgrade slots, so the performance impact would be minimal in normal cases. Best Regards, Hou zj ^ permalink raw reply [nested|flat] 11+ messages in thread
* RE: Fix premature xmin advancement during fast forward decoding 2025-04-22 07:06 Fix premature xmin advancement during fast forward decoding Zhijie Hou (Fujitsu) <[email protected]> @ 2025-04-25 02:54 ` Zhijie Hou (Fujitsu) <[email protected]> 2 siblings, 0 replies; 11+ messages in thread From: Zhijie Hou (Fujitsu) @ 2025-04-25 02:54 UTC (permalink / raw) To: Masahiko Sawada <[email protected]>; +Cc: PostgreSQL Hackers <[email protected]>; Amit Kapila <[email protected]>; shveta malik <[email protected]> On Fri, Apr 25, 2025 at 5:44 AM Masahiko Sawada wrote: (Resending this email after compressing the attachment) > On Tue, Apr 22, 2025 at 12:06 AM Zhijie Hou (Fujitsu) > <[email protected]> wrote: > > > > Hi, > > > > When analyzing some issues in another thread[1], I found a bug that > > fast forward decoding could lead to premature advancement of > > catalog_xmin, resulting in required catalog data being removed by vacuum. > > > > The issue arises because we do not build a base snapshot when decoding > > changes during fast forward mode, preventing reference to the minimum > > transaction ID that remains visible in the snapshot when determining > > the candidate for catalog_xmin. As a result, catalog_xmin was directly > > advanced to the oldest running transaction ID found in the latest > running_xacts record. > > > > In code-level, SnapBuildProcessChange -> ReorderBufferSetBaseSnapshot > > could not be reached during fast forward decoding, resulting in > > rb->txns_by_base_snapshot_lsn being NULL. When advancing > catalog_xmin, > > rb->the > > system attempts to refer to rb->txns_by_base_snapshot_lsn via > > ReorderBufferGetOldestXmin(). However, since > > rb->txns_by_base_snapshot_lsn is NULL, it defaults to directly using > > running->oldestRunningXid as the candidate for catalog_xmin. For > > reference, see the details in SnapBuildProcessRunningXacts. > > I agree with your analysis. > > > To fix this, I think we can allow the base snapshot to be built during > > fast forward decoding, as implemented in the patch 0001 (We already > > built base snapshot in fast-forward mode for logical message in > logicalmsg_decode()). > > > > I also explored the possibility of further optimizing the fix to > > reduce the overhead associated with building a snapshot in > > fast-forward mode. E.g., to maintain a single base_snapshot_xmin > > rather than generating a full snapshot for each transaction, and use > > this base_snapshot_xmin when determining the candidate catalog_xmin. > > However, I am not feeling confident about maintaining some > > fast-forward dedicated fields in common structures and perhaps employing > different handling for catalog_xmin. > > > > Moreover, I conducted a basic test[2] to test the patch's impact, > > noting that advancing the slot incurs roughly a 4% increase in > > processing time after applying the patch, which appears to be > > acceptable. Additionally, the cost associated with building the > > snapshot via SnapBuildBuildSnapshot() did not show up in the profile. > > Where did the regression come from? I think that even with your patch > SnapBuildBuildSnapshot() would be called only a few times in the workload. AFAICS, the primary cost arises from the additional function invocations/executions. Functions such as SnapBuildProcessChange, ReorderBufferSetBaseSnapshot, and ReorderBufferXidHasBaseSnapshot are invoked more frequently after the patch. Although these functions aren't inherently expensive, the scenario I tested involved numerous transactions starting with just a single insert each. This resulted in a high frequency of calls to these functions. Attaching the profile for both HEAD and PATCHED for reference. > With the patch, I think that we need to create ReorderBufferTXN entries for > each transaction even during fast_forward mode, which could lead to > overheads. I think ReorderBufferTXN was created in fast_forward even without the patch. See heap_decode-> ReorderBufferProcessXid. > I think that 4% degradation is something that we want to avoid. As mentioned above, these costs arise from essential function executions required to maintain txns_by_base_snapshot_lsn. So I think the cost is reasonable to ensure correctness. Thoughts ? Best Regards, Hou zj Attachments: [application/x-zip-compressed] profiles.zip (38.0K, ../../OS0PR01MB57164BA5B246184A6DA6438794842@OS0PR01MB5716.jpnprd01.prod.outlook.com/2-profiles.zip) download ^ permalink raw reply [nested|flat] 11+ messages in thread
* Re: Fix premature xmin advancement during fast forward decoding 2025-04-22 07:06 Fix premature xmin advancement during fast forward decoding Zhijie Hou (Fujitsu) <[email protected]> @ 2025-04-26 12:01 ` Amit Kapila <[email protected]> 2025-04-27 06:06 ` Re: Fix premature xmin advancement during fast forward decoding Masahiko Sawada <[email protected]> 2 siblings, 1 reply; 11+ messages in thread From: Amit Kapila @ 2025-04-26 12:01 UTC (permalink / raw) To: Masahiko Sawada <[email protected]>; +Cc: Zhijie Hou (Fujitsu) <[email protected]>; PostgreSQL Hackers <[email protected]>; shveta malik <[email protected]> On Sat, Apr 26, 2025 at 12:01 AM Masahiko Sawada <[email protected]> wrote: > > On Fri, Apr 25, 2025 at 4:42 AM Amit Kapila <[email protected]> wrote: > > > > On Fri, Apr 25, 2025 at 10:46 AM Masahiko Sawada <[email protected]> wrote: > > > > > > What I'm concerned about is the back branches. With this approach all > > > back branches will have such degradations and it doesn't make sense to > > > me to optimize SnapBuildCommitTxn() codes in back branches. > > > > > > > One possibility could be that instead of maintaining an entire > > snapshot in fast_forward mode, we can maintain snapshot's xmin in each > > ReorderBufferTXN. But then also, how would we get the minimum > > txns_by_base_snapshot_lsn as we are getting now in > > ReorderBufferGetOldestXmin? I think we need to traverse the entire > > list of txns to get it in fast_forward mode but that may not show up > > because it will not be done for each transaction. We can try such a > > thing, but it won't be clean to have fast_forward specific code and > > also it would be better to add such things only for HEAD. > > Agreed. > We may need to consider handling xmin in SnapBuildCommitTxn(), where we also set the base snapshot. > > Can you think of any better ideas? > > No idea. Hmm, there seems no reasonable way to fix this issue for back > branches. I consented to the view that these costs were something that > we should have paid from the beginning. > Right, I feel we should go with the simple change proposed by Hou-San for now to fix the bug. If, in the future, we encounter any cases where such optimizations can help for fast-forward mode, then we can consider it. Does that sound reasonable to you? -- With Regards, Amit Kapila. ^ permalink raw reply [nested|flat] 11+ messages in thread
* Re: Fix premature xmin advancement during fast forward decoding 2025-04-22 07:06 Fix premature xmin advancement during fast forward decoding Zhijie Hou (Fujitsu) <[email protected]> 2025-04-26 12:01 ` Re: Fix premature xmin advancement during fast forward decoding Amit Kapila <[email protected]> @ 2025-04-27 06:06 ` Masahiko Sawada <[email protected]> 2025-04-27 07:03 ` RE: Fix premature xmin advancement during fast forward decoding Zhijie Hou (Fujitsu) <[email protected]> 0 siblings, 1 reply; 11+ messages in thread From: Masahiko Sawada @ 2025-04-27 06:06 UTC (permalink / raw) To: Amit Kapila <[email protected]>; +Cc: Zhijie Hou (Fujitsu) <[email protected]>; PostgreSQL Hackers <[email protected]>; shveta malik <[email protected]> On Sat, Apr 26, 2025 at 5:01 AM Amit Kapila <[email protected]> wrote: > > On Sat, Apr 26, 2025 at 12:01 AM Masahiko Sawada <[email protected]> wrote: > > > > On Fri, Apr 25, 2025 at 4:42 AM Amit Kapila <[email protected]> wrote: > > > > > > Can you think of any better ideas? > > > > No idea. Hmm, there seems no reasonable way to fix this issue for back > > branches. I consented to the view that these costs were something that > > we should have paid from the beginning. > > > > Right, I feel we should go with the simple change proposed by Hou-San > for now to fix the bug. If, in the future, we encounter any cases > where such optimizations can help for fast-forward mode, then we can > consider it. Does that sound reasonable to you? Yes, agreed with this approach. Regards, -- Masahiko Sawada Amazon Web Services: https://aws.amazon.com ^ permalink raw reply [nested|flat] 11+ messages in thread
* RE: Fix premature xmin advancement during fast forward decoding 2025-04-22 07:06 Fix premature xmin advancement during fast forward decoding Zhijie Hou (Fujitsu) <[email protected]> 2025-04-26 12:01 ` Re: Fix premature xmin advancement during fast forward decoding Amit Kapila <[email protected]> 2025-04-27 06:06 ` Re: Fix premature xmin advancement during fast forward decoding Masahiko Sawada <[email protected]> @ 2025-04-27 07:03 ` Zhijie Hou (Fujitsu) <[email protected]> 2025-04-28 09:20 ` Re: Fix premature xmin advancement during fast forward decoding Amit Kapila <[email protected]> 0 siblings, 1 reply; 11+ messages in thread From: Zhijie Hou (Fujitsu) @ 2025-04-27 07:03 UTC (permalink / raw) To: Masahiko Sawada <[email protected]>; Amit Kapila <[email protected]>; +Cc: PostgreSQL Hackers <[email protected]>; shveta malik <[email protected]> On Sun, Apr 27, 2025 at 2:07 PM Masahiko Sawada wrote: > > On Sat, Apr 26, 2025 at 5:01 AM Amit Kapila <[email protected]> > wrote: > > > > On Sat, Apr 26, 2025 at 12:01 AM Masahiko Sawada > <[email protected]> wrote: > > > > > > On Fri, Apr 25, 2025 at 4:42 AM Amit Kapila <[email protected]> > wrote: > > > > > > > > Can you think of any better ideas? > > > > > > No idea. Hmm, there seems no reasonable way to fix this issue for > > > back branches. I consented to the view that these costs were > > > something that we should have paid from the beginning. > > > > > > > Right, I feel we should go with the simple change proposed by Hou-San > > for now to fix the bug. If, in the future, we encounter any cases > > where such optimizations can help for fast-forward mode, then we can > > consider it. Does that sound reasonable to you? > > Yes, agreed with this approach. +1. Thanks for the discussion! Here is the V2 patch (including both HEAD and back-branch versions) which merged Amit's suggestions for the comments. It can pass regression and pgindent check. I also adjusted the commit message to mention the commit f49a80c4 as suggested by Amit. Best Regards, Hou zj From 16ae28e3b3b4c10bd82ad679194325eb56aa3965 Mon Sep 17 00:00:00 2001 From: Zhijie Hou <[email protected]> Date: Sun, 27 Apr 2025 14:53:35 +0800 Subject: [PATCH v2] Fix premature xmin advancement during fast forward decoding Currently, fast-forward decoding could prematurely advance catalog_xmin, resulting in required catalog data being removed by vacuum. Commit f49a80c4 fixed a similar issue where the logical slot's catalog_xmin was advanced prematurely during non-fast-forward mode, which involved using the minimum transaction ID remaining visible in the base snapshot when determining candidates for catalog_xmin. However, this fix did not apply to fast-forward mode, as a base snapshot is not built when decoding changes in this mode. Consequently, catalog_xmin was directly advanced to the oldest running transaction ID found in the latest running_xacts record. To resolve this, the commit allows the base snapshot to be built during fast forward decoding. --- .../test_decoding/expected/oldest_xmin.out | 41 +++++++++++++++++++ contrib/test_decoding/specs/oldest_xmin.spec | 5 +++ src/backend/replication/logical/decode.c | 38 +++++++++++------ 3 files changed, 71 insertions(+), 13 deletions(-) diff --git a/contrib/test_decoding/expected/oldest_xmin.out b/contrib/test_decoding/expected/oldest_xmin.out index dd6053f9c1f..57268b38d33 100644 --- a/contrib/test_decoding/expected/oldest_xmin.out +++ b/contrib/test_decoding/expected/oldest_xmin.out @@ -38,3 +38,44 @@ COMMIT stop (1 row) + +starting permutation: s0_begin s0_getxid s1_begin s1_insert s0_alter s0_commit s0_checkpoint s0_advance_slot s0_advance_slot s1_commit s0_vacuum s0_get_changes +step s0_begin: BEGIN; +step s0_getxid: SELECT pg_current_xact_id() IS NULL; +?column? +-------- +f +(1 row) + +step s1_begin: BEGIN; +step s1_insert: INSERT INTO harvest VALUES ((1, 2, 3)); +step s0_alter: ALTER TYPE basket DROP ATTRIBUTE mangos; +step s0_commit: COMMIT; +step s0_checkpoint: CHECKPOINT; +step s0_advance_slot: SELECT slot_name FROM pg_replication_slot_advance('isolation_slot', pg_current_wal_lsn()); +slot_name +-------------- +isolation_slot +(1 row) + +step s0_advance_slot: SELECT slot_name FROM pg_replication_slot_advance('isolation_slot', pg_current_wal_lsn()); +slot_name +-------------- +isolation_slot +(1 row) + +step s1_commit: COMMIT; +step s0_vacuum: VACUUM pg_attribute; +step s0_get_changes: SELECT data FROM pg_logical_slot_get_changes('isolation_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1'); +data +------------------------------------------------------ +BEGIN +table public.harvest: INSERT: fruits[basket]:'(1,2,3)' +COMMIT +(3 rows) + +?column? +-------- +stop +(1 row) + diff --git a/contrib/test_decoding/specs/oldest_xmin.spec b/contrib/test_decoding/specs/oldest_xmin.spec index 88bd30f5ff7..7f2fe3d7ed7 100644 --- a/contrib/test_decoding/specs/oldest_xmin.spec +++ b/contrib/test_decoding/specs/oldest_xmin.spec @@ -25,6 +25,7 @@ step "s0_commit" { COMMIT; } step "s0_checkpoint" { CHECKPOINT; } step "s0_vacuum" { VACUUM pg_attribute; } step "s0_get_changes" { SELECT data FROM pg_logical_slot_get_changes('isolation_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1'); } +step "s0_advance_slot" { SELECT slot_name FROM pg_replication_slot_advance('isolation_slot', pg_current_wal_lsn()); } session "s1" setup { SET synchronous_commit=on; } @@ -40,3 +41,7 @@ step "s1_commit" { COMMIT; } # will be removed (xmax set) before T1 commits. That is, interlocking doesn't # forbid modifying catalog after someone read it (and didn't commit yet). permutation "s0_begin" "s0_getxid" "s1_begin" "s1_insert" "s0_alter" "s0_commit" "s0_checkpoint" "s0_get_changes" "s0_get_changes" "s1_commit" "s0_vacuum" "s0_get_changes" + +# Perform the same testing process as described above, but use advance_slot to +# forces xmin advancement during fast forward decoding. +permutation "s0_begin" "s0_getxid" "s1_begin" "s1_insert" "s0_alter" "s0_commit" "s0_checkpoint" "s0_advance_slot" "s0_advance_slot" "s1_commit" "s0_vacuum" "s0_get_changes" diff --git a/src/backend/replication/logical/decode.c b/src/backend/replication/logical/decode.c index 297eb11b5a8..9c2f1f17c7f 100644 --- a/src/backend/replication/logical/decode.c +++ b/src/backend/replication/logical/decode.c @@ -362,20 +362,24 @@ DecodeHeap2Op(LogicalDecodingContext *ctx, XLogRecordBuffer *buf) /* * If we don't have snapshot or we are just fast-forwarding, there is no - * point in decoding changes. + * point in decoding data changes. However, it's crucial to build the base + * snapshot during fast-forward mode (as is done in + * SnapBuildProcessChange()) because we require the snapshot's xmin when + * determining the candidate catalog_xmin for the replication slot. See + * SnapBuildProcessRunningXacts(). */ - if (SnapBuildCurrentState(builder) < SNAPBUILD_FULL_SNAPSHOT || - ctx->fast_forward) + if (SnapBuildCurrentState(builder) < SNAPBUILD_FULL_SNAPSHOT) return; switch (info) { case XLOG_HEAP2_MULTI_INSERT: - if (!ctx->fast_forward && - SnapBuildProcessChange(builder, xid, buf->origptr)) + if (SnapBuildProcessChange(builder, xid, buf->origptr) && + !ctx->fast_forward) DecodeMultiInsert(ctx, buf); break; case XLOG_HEAP2_NEW_CID: + if (!ctx->fast_forward) { xl_heap_new_cid *xlrec; @@ -422,16 +426,20 @@ DecodeHeapOp(LogicalDecodingContext *ctx, XLogRecordBuffer *buf) /* * If we don't have snapshot or we are just fast-forwarding, there is no - * point in decoding data changes. + * point in decoding data changes. However, it's crucial to build the base + * snapshot during fast-forward mode (as is done in + * SnapBuildProcessChange()) because we require the snapshot's xmin when + * determining the candidate catalog_xmin for the replication slot. See + * SnapBuildProcessRunningXacts(). */ - if (SnapBuildCurrentState(builder) < SNAPBUILD_FULL_SNAPSHOT || - ctx->fast_forward) + if (SnapBuildCurrentState(builder) < SNAPBUILD_FULL_SNAPSHOT) return; switch (info) { case XLOG_HEAP_INSERT: - if (SnapBuildProcessChange(builder, xid, buf->origptr)) + if (SnapBuildProcessChange(builder, xid, buf->origptr) && + !ctx->fast_forward) DecodeInsert(ctx, buf); break; @@ -442,17 +450,20 @@ DecodeHeapOp(LogicalDecodingContext *ctx, XLogRecordBuffer *buf) */ case XLOG_HEAP_HOT_UPDATE: case XLOG_HEAP_UPDATE: - if (SnapBuildProcessChange(builder, xid, buf->origptr)) + if (SnapBuildProcessChange(builder, xid, buf->origptr) && + !ctx->fast_forward) DecodeUpdate(ctx, buf); break; case XLOG_HEAP_DELETE: - if (SnapBuildProcessChange(builder, xid, buf->origptr)) + if (SnapBuildProcessChange(builder, xid, buf->origptr) && + !ctx->fast_forward) DecodeDelete(ctx, buf); break; case XLOG_HEAP_TRUNCATE: - if (SnapBuildProcessChange(builder, xid, buf->origptr)) + if (SnapBuildProcessChange(builder, xid, buf->origptr) && + !ctx->fast_forward) DecodeTruncate(ctx, buf); break; @@ -480,7 +491,8 @@ DecodeHeapOp(LogicalDecodingContext *ctx, XLogRecordBuffer *buf) break; case XLOG_HEAP_CONFIRM: - if (SnapBuildProcessChange(builder, xid, buf->origptr)) + if (SnapBuildProcessChange(builder, xid, buf->origptr) && + !ctx->fast_forward) DecodeSpecConfirm(ctx, buf); break; -- 2.30.0.windows.2 Attachments: [application/octet-stream] v2-0001-HEAD-PG17-Fix-premature-xmin-advancement-during-fast-forwar.patch (7.4K, ../../OS0PR01MB57168059D46306B0A3D4824494862@OS0PR01MB5716.jpnprd01.prod.outlook.com/2-v2-0001-HEAD-PG17-Fix-premature-xmin-advancement-during-fast-forwar.patch) download | inline diff: From 60dde809b654b4f4ae87200cd427372df2fd3521 Mon Sep 17 00:00:00 2001 From: Zhijie Hou <[email protected]> Date: Sun, 27 Apr 2025 14:48:06 +0800 Subject: [PATCH v2] Fix premature xmin advancement during fast forward decoding Currently, fast-forward decoding could prematurely advance catalog_xmin, resulting in required catalog data being removed by vacuum. Commit f49a80c4 fixed a similar issue where the logical slot's catalog_xmin was advanced prematurely during non-fast-forward mode, which involved using the minimum transaction ID remaining visible in the base snapshot when determining candidates for catalog_xmin. However, this fix did not apply to fast-forward mode, as a base snapshot is not built when decoding changes in this mode. Consequently, catalog_xmin was directly advanced to the oldest running transaction ID found in the latest running_xacts record. To resolve this, the commit allows the base snapshot to be built during fast forward decoding. --- .../test_decoding/expected/oldest_xmin.out | 41 +++++++++++++++++++ contrib/test_decoding/specs/oldest_xmin.spec | 5 +++ src/backend/replication/logical/decode.c | 37 +++++++++++------ 3 files changed, 71 insertions(+), 12 deletions(-) diff --git a/contrib/test_decoding/expected/oldest_xmin.out b/contrib/test_decoding/expected/oldest_xmin.out index dd6053f9c1f..57268b38d33 100644 --- a/contrib/test_decoding/expected/oldest_xmin.out +++ b/contrib/test_decoding/expected/oldest_xmin.out @@ -38,3 +38,44 @@ COMMIT stop (1 row) + +starting permutation: s0_begin s0_getxid s1_begin s1_insert s0_alter s0_commit s0_checkpoint s0_advance_slot s0_advance_slot s1_commit s0_vacuum s0_get_changes +step s0_begin: BEGIN; +step s0_getxid: SELECT pg_current_xact_id() IS NULL; +?column? +-------- +f +(1 row) + +step s1_begin: BEGIN; +step s1_insert: INSERT INTO harvest VALUES ((1, 2, 3)); +step s0_alter: ALTER TYPE basket DROP ATTRIBUTE mangos; +step s0_commit: COMMIT; +step s0_checkpoint: CHECKPOINT; +step s0_advance_slot: SELECT slot_name FROM pg_replication_slot_advance('isolation_slot', pg_current_wal_lsn()); +slot_name +-------------- +isolation_slot +(1 row) + +step s0_advance_slot: SELECT slot_name FROM pg_replication_slot_advance('isolation_slot', pg_current_wal_lsn()); +slot_name +-------------- +isolation_slot +(1 row) + +step s1_commit: COMMIT; +step s0_vacuum: VACUUM pg_attribute; +step s0_get_changes: SELECT data FROM pg_logical_slot_get_changes('isolation_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1'); +data +------------------------------------------------------ +BEGIN +table public.harvest: INSERT: fruits[basket]:'(1,2,3)' +COMMIT +(3 rows) + +?column? +-------- +stop +(1 row) + diff --git a/contrib/test_decoding/specs/oldest_xmin.spec b/contrib/test_decoding/specs/oldest_xmin.spec index 88bd30f5ff7..7f2fe3d7ed7 100644 --- a/contrib/test_decoding/specs/oldest_xmin.spec +++ b/contrib/test_decoding/specs/oldest_xmin.spec @@ -25,6 +25,7 @@ step "s0_commit" { COMMIT; } step "s0_checkpoint" { CHECKPOINT; } step "s0_vacuum" { VACUUM pg_attribute; } step "s0_get_changes" { SELECT data FROM pg_logical_slot_get_changes('isolation_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1'); } +step "s0_advance_slot" { SELECT slot_name FROM pg_replication_slot_advance('isolation_slot', pg_current_wal_lsn()); } session "s1" setup { SET synchronous_commit=on; } @@ -40,3 +41,7 @@ step "s1_commit" { COMMIT; } # will be removed (xmax set) before T1 commits. That is, interlocking doesn't # forbid modifying catalog after someone read it (and didn't commit yet). permutation "s0_begin" "s0_getxid" "s1_begin" "s1_insert" "s0_alter" "s0_commit" "s0_checkpoint" "s0_get_changes" "s0_get_changes" "s1_commit" "s0_vacuum" "s0_get_changes" + +# Perform the same testing process as described above, but use advance_slot to +# forces xmin advancement during fast forward decoding. +permutation "s0_begin" "s0_getxid" "s1_begin" "s1_insert" "s0_alter" "s0_commit" "s0_checkpoint" "s0_advance_slot" "s0_advance_slot" "s1_commit" "s0_vacuum" "s0_get_changes" diff --git a/src/backend/replication/logical/decode.c b/src/backend/replication/logical/decode.c index 78f9a0a11c4..cc03f0706e9 100644 --- a/src/backend/replication/logical/decode.c +++ b/src/backend/replication/logical/decode.c @@ -412,19 +412,24 @@ heap2_decode(LogicalDecodingContext *ctx, XLogRecordBuffer *buf) /* * If we don't have snapshot or we are just fast-forwarding, there is no - * point in decoding changes. + * point in decoding data changes. However, it's crucial to build the base + * snapshot during fast-forward mode (as is done in + * SnapBuildProcessChange()) because we require the snapshot's xmin when + * determining the candidate catalog_xmin for the replication slot. See + * SnapBuildProcessRunningXacts(). */ - if (SnapBuildCurrentState(builder) < SNAPBUILD_FULL_SNAPSHOT || - ctx->fast_forward) + if (SnapBuildCurrentState(builder) < SNAPBUILD_FULL_SNAPSHOT) return; switch (info) { case XLOG_HEAP2_MULTI_INSERT: - if (SnapBuildProcessChange(builder, xid, buf->origptr)) + if (SnapBuildProcessChange(builder, xid, buf->origptr) && + !ctx->fast_forward) DecodeMultiInsert(ctx, buf); break; case XLOG_HEAP2_NEW_CID: + if (!ctx->fast_forward) { xl_heap_new_cid *xlrec; @@ -471,16 +476,20 @@ heap_decode(LogicalDecodingContext *ctx, XLogRecordBuffer *buf) /* * If we don't have snapshot or we are just fast-forwarding, there is no - * point in decoding data changes. + * point in decoding data changes. However, it's crucial to build the base + * snapshot during fast-forward mode (as is done in + * SnapBuildProcessChange()) because we require the snapshot's xmin when + * determining the candidate catalog_xmin for the replication slot. See + * SnapBuildProcessRunningXacts(). */ - if (SnapBuildCurrentState(builder) < SNAPBUILD_FULL_SNAPSHOT || - ctx->fast_forward) + if (SnapBuildCurrentState(builder) < SNAPBUILD_FULL_SNAPSHOT) return; switch (info) { case XLOG_HEAP_INSERT: - if (SnapBuildProcessChange(builder, xid, buf->origptr)) + if (SnapBuildProcessChange(builder, xid, buf->origptr) && + !ctx->fast_forward) DecodeInsert(ctx, buf); break; @@ -491,17 +500,20 @@ heap_decode(LogicalDecodingContext *ctx, XLogRecordBuffer *buf) */ case XLOG_HEAP_HOT_UPDATE: case XLOG_HEAP_UPDATE: - if (SnapBuildProcessChange(builder, xid, buf->origptr)) + if (SnapBuildProcessChange(builder, xid, buf->origptr) && + !ctx->fast_forward) DecodeUpdate(ctx, buf); break; case XLOG_HEAP_DELETE: - if (SnapBuildProcessChange(builder, xid, buf->origptr)) + if (SnapBuildProcessChange(builder, xid, buf->origptr) && + !ctx->fast_forward) DecodeDelete(ctx, buf); break; case XLOG_HEAP_TRUNCATE: - if (SnapBuildProcessChange(builder, xid, buf->origptr)) + if (SnapBuildProcessChange(builder, xid, buf->origptr) && + !ctx->fast_forward) DecodeTruncate(ctx, buf); break; @@ -525,7 +537,8 @@ heap_decode(LogicalDecodingContext *ctx, XLogRecordBuffer *buf) break; case XLOG_HEAP_CONFIRM: - if (SnapBuildProcessChange(builder, xid, buf->origptr)) + if (SnapBuildProcessChange(builder, xid, buf->origptr) && + !ctx->fast_forward) DecodeSpecConfirm(ctx, buf); break; -- 2.30.0.windows.2 [text/plain] v2-0001-PG13-16-Fix-premature-xmin-advancement-during-fast-forwar.patch.txt (7.4K, ../../OS0PR01MB57168059D46306B0A3D4824494862@OS0PR01MB5716.jpnprd01.prod.outlook.com/3-v2-0001-PG13-16-Fix-premature-xmin-advancement-during-fast-forwar.patch.txt) download | inline diff: From 16ae28e3b3b4c10bd82ad679194325eb56aa3965 Mon Sep 17 00:00:00 2001 From: Zhijie Hou <[email protected]> Date: Sun, 27 Apr 2025 14:53:35 +0800 Subject: [PATCH v2] Fix premature xmin advancement during fast forward decoding Currently, fast-forward decoding could prematurely advance catalog_xmin, resulting in required catalog data being removed by vacuum. Commit f49a80c4 fixed a similar issue where the logical slot's catalog_xmin was advanced prematurely during non-fast-forward mode, which involved using the minimum transaction ID remaining visible in the base snapshot when determining candidates for catalog_xmin. However, this fix did not apply to fast-forward mode, as a base snapshot is not built when decoding changes in this mode. Consequently, catalog_xmin was directly advanced to the oldest running transaction ID found in the latest running_xacts record. To resolve this, the commit allows the base snapshot to be built during fast forward decoding. --- .../test_decoding/expected/oldest_xmin.out | 41 +++++++++++++++++++ contrib/test_decoding/specs/oldest_xmin.spec | 5 +++ src/backend/replication/logical/decode.c | 38 +++++++++++------ 3 files changed, 71 insertions(+), 13 deletions(-) diff --git a/contrib/test_decoding/expected/oldest_xmin.out b/contrib/test_decoding/expected/oldest_xmin.out index dd6053f9c1f..57268b38d33 100644 --- a/contrib/test_decoding/expected/oldest_xmin.out +++ b/contrib/test_decoding/expected/oldest_xmin.out @@ -38,3 +38,44 @@ COMMIT stop (1 row) + +starting permutation: s0_begin s0_getxid s1_begin s1_insert s0_alter s0_commit s0_checkpoint s0_advance_slot s0_advance_slot s1_commit s0_vacuum s0_get_changes +step s0_begin: BEGIN; +step s0_getxid: SELECT pg_current_xact_id() IS NULL; +?column? +-------- +f +(1 row) + +step s1_begin: BEGIN; +step s1_insert: INSERT INTO harvest VALUES ((1, 2, 3)); +step s0_alter: ALTER TYPE basket DROP ATTRIBUTE mangos; +step s0_commit: COMMIT; +step s0_checkpoint: CHECKPOINT; +step s0_advance_slot: SELECT slot_name FROM pg_replication_slot_advance('isolation_slot', pg_current_wal_lsn()); +slot_name +-------------- +isolation_slot +(1 row) + +step s0_advance_slot: SELECT slot_name FROM pg_replication_slot_advance('isolation_slot', pg_current_wal_lsn()); +slot_name +-------------- +isolation_slot +(1 row) + +step s1_commit: COMMIT; +step s0_vacuum: VACUUM pg_attribute; +step s0_get_changes: SELECT data FROM pg_logical_slot_get_changes('isolation_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1'); +data +------------------------------------------------------ +BEGIN +table public.harvest: INSERT: fruits[basket]:'(1,2,3)' +COMMIT +(3 rows) + +?column? +-------- +stop +(1 row) + diff --git a/contrib/test_decoding/specs/oldest_xmin.spec b/contrib/test_decoding/specs/oldest_xmin.spec index 88bd30f5ff7..7f2fe3d7ed7 100644 --- a/contrib/test_decoding/specs/oldest_xmin.spec +++ b/contrib/test_decoding/specs/oldest_xmin.spec @@ -25,6 +25,7 @@ step "s0_commit" { COMMIT; } step "s0_checkpoint" { CHECKPOINT; } step "s0_vacuum" { VACUUM pg_attribute; } step "s0_get_changes" { SELECT data FROM pg_logical_slot_get_changes('isolation_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1'); } +step "s0_advance_slot" { SELECT slot_name FROM pg_replication_slot_advance('isolation_slot', pg_current_wal_lsn()); } session "s1" setup { SET synchronous_commit=on; } @@ -40,3 +41,7 @@ step "s1_commit" { COMMIT; } # will be removed (xmax set) before T1 commits. That is, interlocking doesn't # forbid modifying catalog after someone read it (and didn't commit yet). permutation "s0_begin" "s0_getxid" "s1_begin" "s1_insert" "s0_alter" "s0_commit" "s0_checkpoint" "s0_get_changes" "s0_get_changes" "s1_commit" "s0_vacuum" "s0_get_changes" + +# Perform the same testing process as described above, but use advance_slot to +# forces xmin advancement during fast forward decoding. +permutation "s0_begin" "s0_getxid" "s1_begin" "s1_insert" "s0_alter" "s0_commit" "s0_checkpoint" "s0_advance_slot" "s0_advance_slot" "s1_commit" "s0_vacuum" "s0_get_changes" diff --git a/src/backend/replication/logical/decode.c b/src/backend/replication/logical/decode.c index 297eb11b5a8..9c2f1f17c7f 100644 --- a/src/backend/replication/logical/decode.c +++ b/src/backend/replication/logical/decode.c @@ -362,20 +362,24 @@ DecodeHeap2Op(LogicalDecodingContext *ctx, XLogRecordBuffer *buf) /* * If we don't have snapshot or we are just fast-forwarding, there is no - * point in decoding changes. + * point in decoding data changes. However, it's crucial to build the base + * snapshot during fast-forward mode (as is done in + * SnapBuildProcessChange()) because we require the snapshot's xmin when + * determining the candidate catalog_xmin for the replication slot. See + * SnapBuildProcessRunningXacts(). */ - if (SnapBuildCurrentState(builder) < SNAPBUILD_FULL_SNAPSHOT || - ctx->fast_forward) + if (SnapBuildCurrentState(builder) < SNAPBUILD_FULL_SNAPSHOT) return; switch (info) { case XLOG_HEAP2_MULTI_INSERT: - if (!ctx->fast_forward && - SnapBuildProcessChange(builder, xid, buf->origptr)) + if (SnapBuildProcessChange(builder, xid, buf->origptr) && + !ctx->fast_forward) DecodeMultiInsert(ctx, buf); break; case XLOG_HEAP2_NEW_CID: + if (!ctx->fast_forward) { xl_heap_new_cid *xlrec; @@ -422,16 +426,20 @@ DecodeHeapOp(LogicalDecodingContext *ctx, XLogRecordBuffer *buf) /* * If we don't have snapshot or we are just fast-forwarding, there is no - * point in decoding data changes. + * point in decoding data changes. However, it's crucial to build the base + * snapshot during fast-forward mode (as is done in + * SnapBuildProcessChange()) because we require the snapshot's xmin when + * determining the candidate catalog_xmin for the replication slot. See + * SnapBuildProcessRunningXacts(). */ - if (SnapBuildCurrentState(builder) < SNAPBUILD_FULL_SNAPSHOT || - ctx->fast_forward) + if (SnapBuildCurrentState(builder) < SNAPBUILD_FULL_SNAPSHOT) return; switch (info) { case XLOG_HEAP_INSERT: - if (SnapBuildProcessChange(builder, xid, buf->origptr)) + if (SnapBuildProcessChange(builder, xid, buf->origptr) && + !ctx->fast_forward) DecodeInsert(ctx, buf); break; @@ -442,17 +450,20 @@ DecodeHeapOp(LogicalDecodingContext *ctx, XLogRecordBuffer *buf) */ case XLOG_HEAP_HOT_UPDATE: case XLOG_HEAP_UPDATE: - if (SnapBuildProcessChange(builder, xid, buf->origptr)) + if (SnapBuildProcessChange(builder, xid, buf->origptr) && + !ctx->fast_forward) DecodeUpdate(ctx, buf); break; case XLOG_HEAP_DELETE: - if (SnapBuildProcessChange(builder, xid, buf->origptr)) + if (SnapBuildProcessChange(builder, xid, buf->origptr) && + !ctx->fast_forward) DecodeDelete(ctx, buf); break; case XLOG_HEAP_TRUNCATE: - if (SnapBuildProcessChange(builder, xid, buf->origptr)) + if (SnapBuildProcessChange(builder, xid, buf->origptr) && + !ctx->fast_forward) DecodeTruncate(ctx, buf); break; @@ -480,7 +491,8 @@ DecodeHeapOp(LogicalDecodingContext *ctx, XLogRecordBuffer *buf) break; case XLOG_HEAP_CONFIRM: - if (SnapBuildProcessChange(builder, xid, buf->origptr)) + if (SnapBuildProcessChange(builder, xid, buf->origptr) && + !ctx->fast_forward) DecodeSpecConfirm(ctx, buf); break; -- 2.30.0.windows.2 ^ permalink raw reply [nested|flat] 11+ messages in thread
* Re: Fix premature xmin advancement during fast forward decoding 2025-04-22 07:06 Fix premature xmin advancement during fast forward decoding Zhijie Hou (Fujitsu) <[email protected]> 2025-04-26 12:01 ` Re: Fix premature xmin advancement during fast forward decoding Amit Kapila <[email protected]> 2025-04-27 06:06 ` Re: Fix premature xmin advancement during fast forward decoding Masahiko Sawada <[email protected]> 2025-04-27 07:03 ` RE: Fix premature xmin advancement during fast forward decoding Zhijie Hou (Fujitsu) <[email protected]> @ 2025-04-28 09:20 ` Amit Kapila <[email protected]> 2025-04-28 16:34 ` Re: Fix premature xmin advancement during fast forward decoding Masahiko Sawada <[email protected]> 0 siblings, 1 reply; 11+ messages in thread From: Amit Kapila @ 2025-04-28 09:20 UTC (permalink / raw) To: Zhijie Hou (Fujitsu) <[email protected]>; +Cc: Masahiko Sawada <[email protected]>; PostgreSQL Hackers <[email protected]>; shveta malik <[email protected]> On Sun, Apr 27, 2025 at 12:33 PM Zhijie Hou (Fujitsu) <[email protected]> wrote: > > I also adjusted the commit message to mention the commit f49a80c4 > as suggested by Amit. > Pushed. -- With Regards, Amit Kapila. ^ permalink raw reply [nested|flat] 11+ messages in thread
* Re: Fix premature xmin advancement during fast forward decoding 2025-04-22 07:06 Fix premature xmin advancement during fast forward decoding Zhijie Hou (Fujitsu) <[email protected]> 2025-04-26 12:01 ` Re: Fix premature xmin advancement during fast forward decoding Amit Kapila <[email protected]> 2025-04-27 06:06 ` Re: Fix premature xmin advancement during fast forward decoding Masahiko Sawada <[email protected]> 2025-04-27 07:03 ` RE: Fix premature xmin advancement during fast forward decoding Zhijie Hou (Fujitsu) <[email protected]> 2025-04-28 09:20 ` Re: Fix premature xmin advancement during fast forward decoding Amit Kapila <[email protected]> @ 2025-04-28 16:34 ` Masahiko Sawada <[email protected]> 0 siblings, 0 replies; 11+ messages in thread From: Masahiko Sawada @ 2025-04-28 16:34 UTC (permalink / raw) To: Amit Kapila <[email protected]>; +Cc: Zhijie Hou (Fujitsu) <[email protected]>; PostgreSQL Hackers <[email protected]>; shveta malik <[email protected]> On Mon, Apr 28, 2025 at 2:20 AM Amit Kapila <[email protected]> wrote: > > On Sun, Apr 27, 2025 at 12:33 PM Zhijie Hou (Fujitsu) > <[email protected]> wrote: > > > > I also adjusted the commit message to mention the commit f49a80c4 > > as suggested by Amit. > > > > Pushed. Thank you! Regards, -- Masahiko Sawada Amazon Web Services: https://aws.amazon.com ^ permalink raw reply [nested|flat] 11+ messages in thread
end of thread, other threads:[~2025-04-28 16:34 UTC | newest] Thread overview: 11+ messages (download: mbox mbox.gz follow: Atom feed) -- links below jump to the message on this page -- 2019-01-22 21:00 [PATCH 1/2] index_get_partition Alvaro Herrera <[email protected]> 2023-05-31 11:46 [PATCH v30 08/11] Add aggregates support in IVM Yugo Nagata <[email protected]> 2025-04-22 07:06 Fix premature xmin advancement during fast forward decoding Zhijie Hou (Fujitsu) <[email protected]> 2025-04-23 06:31 ` Re: Fix premature xmin advancement during fast forward decoding shveta malik <[email protected]> 2025-04-23 08:20 ` RE: Fix premature xmin advancement during fast forward decoding Zhijie Hou (Fujitsu) <[email protected]> 2025-04-25 02:54 ` RE: Fix premature xmin advancement during fast forward decoding Zhijie Hou (Fujitsu) <[email protected]> 2025-04-26 12:01 ` Re: Fix premature xmin advancement during fast forward decoding Amit Kapila <[email protected]> 2025-04-27 06:06 ` Re: Fix premature xmin advancement during fast forward decoding Masahiko Sawada <[email protected]> 2025-04-27 07:03 ` RE: Fix premature xmin advancement during fast forward decoding Zhijie Hou (Fujitsu) <[email protected]> 2025-04-28 09:20 ` Re: Fix premature xmin advancement during fast forward decoding Amit Kapila <[email protected]> 2025-04-28 16:34 ` Re: Fix premature xmin advancement during fast forward decoding Masahiko Sawada <[email protected]>
This inbox is served by agora; see mirroring instructions for how to clone and mirror all data and code used for this inbox