Commit f63a5ead authored by Tom Lane's avatar Tom Lane

Avoid touching replica identity index in ExtractReplicaIdentity().

In what seems like a fit of misplaced optimization,
ExtractReplicaIdentity() accessed the relation's replica-identity
index without taking any lock on it.  Usually, the surrounding query
already holds some lock so this is safe enough ... but in the case
of a previously-planned delete, there might be no existing lock.
Given a suitable test case, this is exposed in v12 and HEAD by an
assertion added by commit b04aeb0a.

The whole thing's rather poorly thought out anyway; rather than
looking directly at the index, we should use the index-attributes
bitmap that's held by the parent table's relcache entry, as the
caller functions do.  This is more consistent and likely a bit
faster, since it avoids a cache lookup.  Hence, change to doing it
that way.

While at it, rather than blithely assuming that the identity
columns are non-null (with catastrophic results if that's wrong),
add assertion checks that they aren't null.  Possibly those should
be actual test-and-elog, but I'll leave it like this for now.

In principle, this is a bug that's been there since this code was
introduced (in 9.4).  In practice, the risk seems quite low, since
we do have a lock on the index's parent table, so concurrent
changes to the index's catalog entries seem unlikely.  Given the
precedent that commit 9c703c16 wasn't back-patched, I won't risk
back-patching this further than v12.

Per report from Hadi Moshayedi.

Discussion: https://postgr.es/m/CAK=1=Wrek44Ese1V7LjKiQS-Nd-5LgLi_5_CskGbpggKEf3tKQ@mail.gmail.com
parent aef36238
...@@ -7593,19 +7593,24 @@ log_heap_new_cid(Relation relation, HeapTuple tup) ...@@ -7593,19 +7593,24 @@ log_heap_new_cid(Relation relation, HeapTuple tup)
* the old tuple in a UPDATE or DELETE. * the old tuple in a UPDATE or DELETE.
* *
* Returns NULL if there's no need to log an identity or if there's no suitable * Returns NULL if there's no need to log an identity or if there's no suitable
* key in the Relation relation. * key defined.
*
* key_changed should be false if caller knows that no replica identity
* columns changed value. It's always true in the DELETE case.
*
* *copy is set to true if the returned tuple is a modified copy rather than
* the same tuple that was passed in.
*/ */
static HeapTuple static HeapTuple
ExtractReplicaIdentity(Relation relation, HeapTuple tp, bool key_changed, bool *copy) ExtractReplicaIdentity(Relation relation, HeapTuple tp, bool key_changed,
bool *copy)
{ {
TupleDesc desc = RelationGetDescr(relation); TupleDesc desc = RelationGetDescr(relation);
Oid replidindex;
Relation idx_rel;
char replident = relation->rd_rel->relreplident; char replident = relation->rd_rel->relreplident;
HeapTuple key_tuple = NULL; Bitmapset *idattrs;
HeapTuple key_tuple;
bool nulls[MaxHeapAttributeNumber]; bool nulls[MaxHeapAttributeNumber];
Datum values[MaxHeapAttributeNumber]; Datum values[MaxHeapAttributeNumber];
int natt;
*copy = false; *copy = false;
...@@ -7624,7 +7629,7 @@ ExtractReplicaIdentity(Relation relation, HeapTuple tp, bool key_changed, bool * ...@@ -7624,7 +7629,7 @@ ExtractReplicaIdentity(Relation relation, HeapTuple tp, bool key_changed, bool *
if (HeapTupleHasExternal(tp)) if (HeapTupleHasExternal(tp))
{ {
*copy = true; *copy = true;
tp = toast_flatten_tuple(tp, RelationGetDescr(relation)); tp = toast_flatten_tuple(tp, desc);
} }
return tp; return tp;
} }
...@@ -7633,41 +7638,39 @@ ExtractReplicaIdentity(Relation relation, HeapTuple tp, bool key_changed, bool * ...@@ -7633,41 +7638,39 @@ ExtractReplicaIdentity(Relation relation, HeapTuple tp, bool key_changed, bool *
if (!key_changed) if (!key_changed)
return NULL; return NULL;
/* find the replica identity index */ /* find out the replica identity columns */
replidindex = RelationGetReplicaIndex(relation); idattrs = RelationGetIndexAttrBitmap(relation,
if (!OidIsValid(replidindex)) INDEX_ATTR_BITMAP_IDENTITY_KEY);
{
elog(DEBUG4, "could not find configured replica identity for table \"%s\"",
RelationGetRelationName(relation));
return NULL;
}
idx_rel = RelationIdGetRelation(replidindex);
Assert(CheckRelationLockedByMe(idx_rel, AccessShareLock, true));
/* deform tuple, so we have fast access to columns */
heap_deform_tuple(tp, desc, values, nulls);
/* set all columns to NULL, regardless of whether they actually are */ /*
memset(nulls, 1, sizeof(nulls)); * If there's no defined replica identity columns, treat as !key_changed.
* (This case should not be reachable from heap_update, since that should
* calculate key_changed accurately. But heap_delete just passes constant
* true for key_changed, so we can hit this case in deletes.)
*/
if (bms_is_empty(idattrs))
return NULL;
/* /*
* Now set all columns contained in the index to NOT NULL, they cannot * Construct a new tuple containing only the replica identity columns,
* currently be NULL. * with nulls elsewhere. While we're at it, assert that the replica
* identity columns aren't null.
*/ */
for (natt = 0; natt < IndexRelationGetNumberOfKeyAttributes(idx_rel); natt++) heap_deform_tuple(tp, desc, values, nulls);
{
int attno = idx_rel->rd_index->indkey.values[natt];
if (attno < 0) for (int i = 0; i < desc->natts; i++)
elog(ERROR, "system column in index"); {
nulls[attno - 1] = false; if (bms_is_member(i + 1 - FirstLowInvalidHeapAttributeNumber,
idattrs))
Assert(!nulls[i]);
else
nulls[i] = true;
} }
key_tuple = heap_form_tuple(desc, values, nulls); key_tuple = heap_form_tuple(desc, values, nulls);
*copy = true; *copy = true;
RelationClose(idx_rel);
bms_free(idattrs);
/* /*
* If the tuple, which by here only contains indexed columns, still has * If the tuple, which by here only contains indexed columns, still has
...@@ -7680,7 +7683,7 @@ ExtractReplicaIdentity(Relation relation, HeapTuple tp, bool key_changed, bool * ...@@ -7680,7 +7683,7 @@ ExtractReplicaIdentity(Relation relation, HeapTuple tp, bool key_changed, bool *
{ {
HeapTuple oldtup = key_tuple; HeapTuple oldtup = key_tuple;
key_tuple = toast_flatten_tuple(oldtup, RelationGetDescr(relation)); key_tuple = toast_flatten_tuple(oldtup, desc);
heap_freetuple(oldtup); heap_freetuple(oldtup);
} }
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment