From 45866c75507f0757de0da6e90c694a0dbe67d727 Mon Sep 17 00:00:00 2001
From: Robert Haas <rhaas@postgresql.org>
Date: Fri, 6 Oct 2017 15:27:11 -0400
Subject: [PATCH] Copy information from the relcache instead of pointing to it.
We have the relations continuously locked, but not open, so relcache
pointers are not guaranteed to be stable. Per buildfarm member
prion.
Ashutosh Bapat. I fixed a typo.
Discussion: http://postgr.es/m/CAFjFpRcRBqoKLZSNmRsjKr81uEP=ennvqSQaXVCCBTXvJ2rW+Q@mail.gmail.com
---
src/backend/catalog/partition.c | 68 ++++++++++++++++++++++++++++
src/backend/optimizer/util/plancat.c | 35 ++++++++++----
src/include/catalog/partition.h | 2 +
3 files changed, 96 insertions(+), 9 deletions(-)
diff --git a/src/backend/catalog/partition.c b/src/backend/catalog/partition.c
index 3a8306a0552..ebda85e4efc 100644
--- a/src/backend/catalog/partition.c
+++ b/src/backend/catalog/partition.c
@@ -701,6 +701,74 @@ partition_bounds_equal(int partnatts, int16 *parttyplen, bool *parttypbyval,
return true;
}
+/*
+ * Return a copy of given PartitionBoundInfo structure. The data types of bounds
+ * are described by given partition key specificiation.
+ */
+extern PartitionBoundInfo
+partition_bounds_copy(PartitionBoundInfo src,
+ PartitionKey key)
+{
+ PartitionBoundInfo dest;
+ int i;
+ int ndatums;
+ int partnatts;
+ int num_indexes;
+
+ dest = (PartitionBoundInfo) palloc(sizeof(PartitionBoundInfoData));
+
+ dest->strategy = src->strategy;
+ ndatums = dest->ndatums = src->ndatums;
+ partnatts = key->partnatts;
+
+ /* Range partitioned table has an extra index. */
+ num_indexes = key->strategy == PARTITION_STRATEGY_RANGE ? ndatums + 1 : ndatums;
+
+ /* List partitioned tables have only a single partition key. */
+ Assert(key->strategy != PARTITION_STRATEGY_LIST || partnatts == 1);
+
+ dest->datums = (Datum **) palloc(sizeof(Datum *) * ndatums);
+
+ if (src->kind != NULL)
+ {
+ dest->kind = (PartitionRangeDatumKind **) palloc(ndatums *
+ sizeof(PartitionRangeDatumKind *));
+ for (i = 0; i < ndatums; i++)
+ {
+ dest->kind[i] = (PartitionRangeDatumKind *) palloc(partnatts *
+ sizeof(PartitionRangeDatumKind));
+
+ memcpy(dest->kind[i], src->kind[i],
+ sizeof(PartitionRangeDatumKind) * key->partnatts);
+ }
+ }
+ else
+ dest->kind = NULL;
+
+ for (i = 0; i < ndatums; i++)
+ {
+ int j;
+ dest->datums[i] = (Datum *) palloc(sizeof(Datum) * partnatts);
+
+ for (j = 0; j < partnatts; j++)
+ {
+ if (dest->kind == NULL ||
+ dest->kind[i][j] == PARTITION_RANGE_DATUM_VALUE)
+ dest->datums[i][j] = datumCopy(src->datums[i][j],
+ key->parttypbyval[j],
+ key->parttyplen[j]);
+ }
+ }
+
+ dest->indexes = (int *) palloc(sizeof(int) * num_indexes);
+ memcpy(dest->indexes, src->indexes, sizeof(int) * num_indexes);
+
+ dest->null_index = src->null_index;
+ dest->default_index = src->default_index;
+
+ return dest;
+}
+
/*
* check_new_partition_bound
*
diff --git a/src/backend/optimizer/util/plancat.c b/src/backend/optimizer/util/plancat.c
index 93cc7576a06..9d35a41e221 100644
--- a/src/backend/optimizer/util/plancat.c
+++ b/src/backend/optimizer/util/plancat.c
@@ -1825,13 +1825,15 @@ set_relation_partition_info(PlannerInfo *root, RelOptInfo *rel,
Relation relation)
{
PartitionDesc partdesc;
+ PartitionKey partkey;
Assert(relation->rd_rel->relkind == RELKIND_PARTITIONED_TABLE);
partdesc = RelationGetPartitionDesc(relation);
+ partkey = RelationGetPartitionKey(relation);
rel->part_scheme = find_partition_scheme(root, relation);
Assert(partdesc != NULL && rel->part_scheme != NULL);
- rel->boundinfo = partdesc->boundinfo;
+ rel->boundinfo = partition_bounds_copy(partdesc->boundinfo, partkey);
rel->nparts = partdesc->nparts;
set_baserel_partition_key_exprs(relation, rel);
}
@@ -1888,18 +1890,33 @@ find_partition_scheme(PlannerInfo *root, Relation relation)
/*
* Did not find matching partition scheme. Create one copying relevant
- * information from the relcache. Instead of copying whole arrays, copy
- * the pointers in relcache. It's safe to do so since
- * RelationClearRelation() wouldn't change it while planner is using it.
+ * information from the relcache. We need to copy the contents of the array
+ * since the relcache entry may not survive after we have closed the
+ * relation.
*/
part_scheme = (PartitionScheme) palloc0(sizeof(PartitionSchemeData));
part_scheme->strategy = partkey->strategy;
part_scheme->partnatts = partkey->partnatts;
- part_scheme->partopfamily = partkey->partopfamily;
- part_scheme->partopcintype = partkey->partopcintype;
- part_scheme->parttypcoll = partkey->parttypcoll;
- part_scheme->parttyplen = partkey->parttyplen;
- part_scheme->parttypbyval = partkey->parttypbyval;
+
+ part_scheme->partopfamily = (Oid *) palloc(sizeof(Oid) * partnatts);
+ memcpy(part_scheme->partopfamily, partkey->partopfamily,
+ sizeof(Oid) * partnatts);
+
+ part_scheme->partopcintype = (Oid *) palloc(sizeof(Oid) * partnatts);
+ memcpy(part_scheme->partopcintype, partkey->partopcintype,
+ sizeof(Oid) * partnatts);
+
+ part_scheme->parttypcoll = (Oid *) palloc(sizeof(Oid) * partnatts);
+ memcpy(part_scheme->parttypcoll, partkey->parttypcoll,
+ sizeof(Oid) * partnatts);
+
+ part_scheme->parttyplen = (int16 *) palloc(sizeof(int16) * partnatts);
+ memcpy(part_scheme->parttyplen, partkey->parttyplen,
+ sizeof(int16) * partnatts);
+
+ part_scheme->parttypbyval = (bool *) palloc(sizeof(bool) * partnatts);
+ memcpy(part_scheme->parttypbyval, partkey->parttypbyval,
+ sizeof(bool) * partnatts);
/* Add the partitioning scheme to PlannerInfo. */
root->part_schemes = lappend(root->part_schemes, part_scheme);
diff --git a/src/include/catalog/partition.h b/src/include/catalog/partition.h
index 454a940a23b..945ac0239d2 100644
--- a/src/include/catalog/partition.h
+++ b/src/include/catalog/partition.h
@@ -74,6 +74,8 @@ extern void RelationBuildPartitionDesc(Relation relation);
extern bool partition_bounds_equal(int partnatts, int16 *parttyplen,
bool *parttypbyval, PartitionBoundInfo b1,
PartitionBoundInfo b2);
+extern PartitionBoundInfo partition_bounds_copy(PartitionBoundInfo src,
+ PartitionKey key);
extern void check_new_partition_bound(char *relname, Relation parent,
PartitionBoundSpec *spec);