1
0
mirror of https://github.com/postgres/postgres.git synced 2025-04-25 21:42:33 +03:00

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
This commit is contained in:
Robert Haas 2017-10-06 15:27:11 -04:00
parent a1c2c430d3
commit 45866c7550
3 changed files with 96 additions and 9 deletions

View File

@ -701,6 +701,74 @@ partition_bounds_equal(int partnatts, int16 *parttyplen, bool *parttypbyval,
return true; 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 * check_new_partition_bound
* *

View File

@ -1825,13 +1825,15 @@ set_relation_partition_info(PlannerInfo *root, RelOptInfo *rel,
Relation relation) Relation relation)
{ {
PartitionDesc partdesc; PartitionDesc partdesc;
PartitionKey partkey;
Assert(relation->rd_rel->relkind == RELKIND_PARTITIONED_TABLE); Assert(relation->rd_rel->relkind == RELKIND_PARTITIONED_TABLE);
partdesc = RelationGetPartitionDesc(relation); partdesc = RelationGetPartitionDesc(relation);
partkey = RelationGetPartitionKey(relation);
rel->part_scheme = find_partition_scheme(root, relation); rel->part_scheme = find_partition_scheme(root, relation);
Assert(partdesc != NULL && rel->part_scheme != NULL); Assert(partdesc != NULL && rel->part_scheme != NULL);
rel->boundinfo = partdesc->boundinfo; rel->boundinfo = partition_bounds_copy(partdesc->boundinfo, partkey);
rel->nparts = partdesc->nparts; rel->nparts = partdesc->nparts;
set_baserel_partition_key_exprs(relation, rel); 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 * Did not find matching partition scheme. Create one copying relevant
* information from the relcache. Instead of copying whole arrays, copy * information from the relcache. We need to copy the contents of the array
* the pointers in relcache. It's safe to do so since * since the relcache entry may not survive after we have closed the
* RelationClearRelation() wouldn't change it while planner is using it. * relation.
*/ */
part_scheme = (PartitionScheme) palloc0(sizeof(PartitionSchemeData)); part_scheme = (PartitionScheme) palloc0(sizeof(PartitionSchemeData));
part_scheme->strategy = partkey->strategy; part_scheme->strategy = partkey->strategy;
part_scheme->partnatts = partkey->partnatts; part_scheme->partnatts = partkey->partnatts;
part_scheme->partopfamily = partkey->partopfamily;
part_scheme->partopcintype = partkey->partopcintype; part_scheme->partopfamily = (Oid *) palloc(sizeof(Oid) * partnatts);
part_scheme->parttypcoll = partkey->parttypcoll; memcpy(part_scheme->partopfamily, partkey->partopfamily,
part_scheme->parttyplen = partkey->parttyplen; sizeof(Oid) * partnatts);
part_scheme->parttypbyval = partkey->parttypbyval;
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. */ /* Add the partitioning scheme to PlannerInfo. */
root->part_schemes = lappend(root->part_schemes, part_scheme); root->part_schemes = lappend(root->part_schemes, part_scheme);

View File

@ -74,6 +74,8 @@ extern void RelationBuildPartitionDesc(Relation relation);
extern bool partition_bounds_equal(int partnatts, int16 *parttyplen, extern bool partition_bounds_equal(int partnatts, int16 *parttyplen,
bool *parttypbyval, PartitionBoundInfo b1, bool *parttypbyval, PartitionBoundInfo b1,
PartitionBoundInfo b2); PartitionBoundInfo b2);
extern PartitionBoundInfo partition_bounds_copy(PartitionBoundInfo src,
PartitionKey key);
extern void check_new_partition_bound(char *relname, Relation parent, extern void check_new_partition_bound(char *relname, Relation parent,
PartitionBoundSpec *spec); PartitionBoundSpec *spec);