public inbox for [email protected]
help / color / mirror / Atom feed[PATCH v1 2/2] ALTER TABLE ... DETACH CONCURRENTLY
5+ messages / 5 participants
[nested] [flat]
* [PATCH v1 2/2] ALTER TABLE ... DETACH CONCURRENTLY
@ 2020-08-03 23:21 Alvaro Herrera <[email protected]>
0 siblings, 0 replies; 5+ messages in thread
From: Alvaro Herrera @ 2020-08-03 23:21 UTC (permalink / raw)
---
doc/src/sgml/catalogs.sgml | 10 +
doc/src/sgml/ref/alter_table.sgml | 25 +-
src/backend/catalog/heap.c | 13 +-
src/backend/catalog/index.c | 4 +-
src/backend/catalog/pg_inherits.c | 51 +-
src/backend/commands/indexcmds.c | 5 +-
src/backend/commands/tablecmds.c | 491 ++++++++++++++----
src/backend/commands/trigger.c | 7 +-
src/backend/executor/execPartition.c | 16 +-
src/backend/nodes/copyfuncs.c | 1 +
src/backend/nodes/equalfuncs.c | 1 +
src/backend/optimizer/util/plancat.c | 2 +-
src/backend/parser/gram.y | 22 +-
src/backend/partitioning/partbounds.c | 6 +-
src/backend/partitioning/partdesc.c | 21 +-
src/backend/tcop/utility.c | 19 +
src/bin/psql/describe.c | 40 +-
src/include/catalog/pg_inherits.h | 7 +-
src/include/nodes/parsenodes.h | 2 +
src/include/parser/kwlist.h | 1 +
src/include/partitioning/partdesc.h | 5 +-
.../detach-partition-concurrently-1.out | 182 +++++++
.../detach-partition-concurrently-2.out | 66 +++
src/test/isolation/isolation_schedule | 2 +
.../detach-partition-concurrently-1.spec | 71 +++
.../detach-partition-concurrently-2.spec | 41 ++
26 files changed, 958 insertions(+), 153 deletions(-)
create mode 100644 src/test/isolation/expected/detach-partition-concurrently-1.out
create mode 100644 src/test/isolation/expected/detach-partition-concurrently-2.out
create mode 100644 src/test/isolation/specs/detach-partition-concurrently-1.spec
create mode 100644 src/test/isolation/specs/detach-partition-concurrently-2.spec
diff --git a/doc/src/sgml/catalogs.sgml b/doc/src/sgml/catalogs.sgml
index 26fda20d19..0f8abdc2d0 100644
--- a/doc/src/sgml/catalogs.sgml
+++ b/doc/src/sgml/catalogs.sgml
@@ -4471,6 +4471,16 @@ SCRAM-SHA-256$<replaceable><iteration count></replaceable>:<replaceable>&l
when using declarative partitioning.
</para></entry>
</row>
+
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>inhdetached</structfield> <type>bool</type>
+ </para>
+ <para>
+ Set to true for a partition that is in the process of being detached;
+ false otherwise.
+ </para></entry>
+ </row>
</tbody>
</tgroup>
</table>
diff --git a/doc/src/sgml/ref/alter_table.sgml b/doc/src/sgml/ref/alter_table.sgml
index b2eb7097a9..2e67468e46 100644
--- a/doc/src/sgml/ref/alter_table.sgml
+++ b/doc/src/sgml/ref/alter_table.sgml
@@ -36,7 +36,9 @@ ALTER TABLE ALL IN TABLESPACE <replaceable class="parameter">name</replaceable>
ALTER TABLE [ IF EXISTS ] <replaceable class="parameter">name</replaceable>
ATTACH PARTITION <replaceable class="parameter">partition_name</replaceable> { FOR VALUES <replaceable class="parameter">partition_bound_spec</replaceable> | DEFAULT }
ALTER TABLE [ IF EXISTS ] <replaceable class="parameter">name</replaceable>
- DETACH PARTITION <replaceable class="parameter">partition_name</replaceable>
+ DETACH PARTITION <replaceable class="parameter">partition_name</replaceable> [ CONCURRENTLY ]
+ALTER TABLE [ IF EXISTS ] <replaceable class="parameter">name</replaceable>
+ DETACH PARTITION <replaceable class="parameter">partition_name</replaceable> FINALIZE
<phrase>where <replaceable class="parameter">action</replaceable> is one of:</phrase>
@@ -938,7 +940,8 @@ WITH ( MODULUS <replaceable class="parameter">numeric_literal</replaceable>, REM
</varlistentry>
<varlistentry>
- <term><literal>DETACH PARTITION</literal> <replaceable class="parameter">partition_name</replaceable></term>
+ <term><literal>DETACH PARTITION</literal> <replaceable class="parameter">partition_name</replaceable> [ { CONCURRENTLY | FINALIZE } ]</term>
+
<listitem>
<para>
This form detaches the specified partition of the target table. The detached
@@ -947,6 +950,24 @@ WITH ( MODULUS <replaceable class="parameter">numeric_literal</replaceable>, REM
attached to the target table's indexes are detached. Any triggers that
were created as clones of those in the target table are removed.
</para>
+ <para>
+ If <literal>CONCURRENTLY</literal> is specified, this process runs in two
+ transactions in order to avoid blocking other sessions that might be accessing
+ the partitioned table. During the first transaction,
+ <literal>SHARE UPDATE EXCLUSIVE</literal> is taken in both parent table and
+ partition, and the partition is marked detached; at that point, the transaction
+ is committed and all transactions using the partitioned table are waited for.
+ Once all those transactions are gone, the second stage acquires
+ <literal>ACCESS EXCLUSIVE</literal> on the partition, and the detach process
+ completes.
+ <literal>CONCURRENTLY</literal> is not allowed if the
+ partitioned table contains a default partition.
+ </para>
+ <para>
+ If <literal>FINALIZE</literal> is specified, complete actions of a
+ previous <literal>DETACH CONCURRENTLY</literal> invocation that
+ was cancelled or crashed.
+ </para>
</listitem>
</varlistentry>
diff --git a/src/backend/catalog/heap.c b/src/backend/catalog/heap.c
index f2ca686397..7cc5cb4d51 100644
--- a/src/backend/catalog/heap.c
+++ b/src/backend/catalog/heap.c
@@ -2552,10 +2552,12 @@ StoreConstraints(Relation rel, List *cooked_constraints, bool is_internal)
* Returns a list of CookedConstraint nodes that shows the cooked form of
* the default and constraint expressions added to the relation.
*
- * NB: caller should have opened rel with AccessExclusiveLock, and should
- * hold that lock till end of transaction. Also, we assume the caller has
- * done a CommandCounterIncrement if necessary to make the relation's catalog
- * tuples visible.
+ * NB: caller should have opened rel with some self-conflicting lock mode,
+ * and should hold that lock till end of transaction; for normal cases that'll
+ * be AccessExclusiveLock, but if caller knows that the constraint is already
+ * enforced by some other means, it can be ShareUpdateExclusiveLock. Also, we
+ * assume the caller has done a CommandCounterIncrement if necessary to make
+ * the relation's catalog tuples visible.
*/
List *
AddRelationNewConstraints(Relation rel,
@@ -3787,7 +3789,8 @@ StorePartitionBound(Relation rel, Relation parent, PartitionBoundSpec *bound)
* relcache entry for that partition every time a partition is added or
* removed.
*/
- defaultPartOid = get_default_oid_from_partdesc(RelationGetPartitionDesc(parent));
+ defaultPartOid =
+ get_default_oid_from_partdesc(RelationGetPartitionDesc(parent, false));
if (OidIsValid(defaultPartOid))
CacheInvalidateRelcacheByRelid(defaultPartOid);
diff --git a/src/backend/catalog/index.c b/src/backend/catalog/index.c
index 1be27eec52..ad545a7866 100644
--- a/src/backend/catalog/index.c
+++ b/src/backend/catalog/index.c
@@ -1655,7 +1655,7 @@ index_concurrently_swap(Oid newIndexId, Oid oldIndexId, const char *oldName)
List *ancestors = get_partition_ancestors(oldIndexId);
Oid parentIndexRelid = linitial_oid(ancestors);
- DeleteInheritsTuple(oldIndexId, parentIndexRelid);
+ DeleteInheritsTuple(oldIndexId, parentIndexRelid, false, NULL);
StoreSingleInheritance(newIndexId, parentIndexRelid, 1);
list_free(ancestors);
@@ -2241,7 +2241,7 @@ index_drop(Oid indexId, bool concurrent, bool concurrent_lock_mode)
/*
* fix INHERITS relation
*/
- DeleteInheritsTuple(indexId, InvalidOid);
+ DeleteInheritsTuple(indexId, InvalidOid, false, NULL);
/*
* We are presently too lazy to attempt to compute the new correct value
diff --git a/src/backend/catalog/pg_inherits.c b/src/backend/catalog/pg_inherits.c
index 17f37eb39f..f143db0ad4 100644
--- a/src/backend/catalog/pg_inherits.c
+++ b/src/backend/catalog/pg_inherits.c
@@ -52,7 +52,8 @@ typedef struct SeenRelsEntry
* against possible DROPs of child relations.
*/
List *
-find_inheritance_children(Oid parentrelId, LOCKMODE lockmode)
+find_inheritance_children(Oid parentrelId, bool include_detached,
+ LOCKMODE lockmode)
{
List *list = NIL;
Relation relation;
@@ -91,7 +92,13 @@ find_inheritance_children(Oid parentrelId, LOCKMODE lockmode)
while ((inheritsTuple = systable_getnext(scan)) != NULL)
{
+ /* skip detached at caller's request */
+ if (((Form_pg_inherits) GETSTRUCT(inheritsTuple))->inhdetached &&
+ !include_detached)
+ continue;
+
inhrelid = ((Form_pg_inherits) GETSTRUCT(inheritsTuple))->inhrelid;
+
if (numoids >= maxoids)
{
maxoids *= 2;
@@ -160,6 +167,9 @@ find_inheritance_children(Oid parentrelId, LOCKMODE lockmode)
* given rel; caller should already have locked it). If lockmode is NoLock
* then no locks are acquired, but caller must beware of race conditions
* against possible DROPs of child relations.
+ *
+ * NB - No current callers of this routine are interested in children being
+ * concurrently detached, so there's no provision to include them.
*/
List *
find_all_inheritors(Oid parentrelId, LOCKMODE lockmode, List **numparents)
@@ -200,7 +210,8 @@ find_all_inheritors(Oid parentrelId, LOCKMODE lockmode, List **numparents)
ListCell *lc;
/* Get the direct children of this rel */
- currentchildren = find_inheritance_children(currentrel, lockmode);
+ currentchildren = find_inheritance_children(currentrel, false,
+ lockmode);
/*
* Add to the queue only those children not already seen. This avoids
@@ -429,6 +440,7 @@ StoreSingleInheritance(Oid relationId, Oid parentOid, int32 seqNumber)
values[Anum_pg_inherits_inhrelid - 1] = ObjectIdGetDatum(relationId);
values[Anum_pg_inherits_inhparent - 1] = ObjectIdGetDatum(parentOid);
values[Anum_pg_inherits_inhseqno - 1] = Int32GetDatum(seqNumber);
+ values[Anum_pg_inherits_inhdetached - 1] = BoolGetDatum(false);
memset(nulls, 0, sizeof(nulls));
@@ -448,10 +460,21 @@ StoreSingleInheritance(Oid relationId, Oid parentOid, int32 seqNumber)
* as InvalidOid, in which case all tuples matching inhrelid are deleted;
* otherwise only delete tuples with the specified inhparent.
*
+ * 'detached' is the expected state of the inhdetached flag. If the catalog
+ * row does not match that state, an error is raised. When used on
+ * partitions, the partition name must be passed, for possible error messages.
+ *
+ * If allow_detached is passed false, then an error is raised if the child is
+ * already marked detached. A name must be supplied in that case, for the
+ * error message. This should only be used with table partitions.
+ * XXX the name bit is pretty weird and problematic for non-partition cases;
+ * what if the flag check fails?
+ *
* Returns whether at least one row was deleted.
*/
bool
-DeleteInheritsTuple(Oid inhrelid, Oid inhparent)
+DeleteInheritsTuple(Oid inhrelid, Oid inhparent, bool detached,
+ const char *childname)
{
bool found = false;
Relation catalogRelation;
@@ -478,6 +501,28 @@ DeleteInheritsTuple(Oid inhrelid, Oid inhparent)
parent = ((Form_pg_inherits) GETSTRUCT(inheritsTuple))->inhparent;
if (!OidIsValid(inhparent) || parent == inhparent)
{
+ bool inhdetached;
+
+ inhdetached = ((Form_pg_inherits) GETSTRUCT(inheritsTuple))->inhdetached;
+
+ /*
+ * Raise error depending on state. This should only happen for
+ * partitions, but we have no way to cross-check.
+ */
+ if (inhdetached && !detached)
+ ereport(ERROR,
+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("cannot detach partition \"%s\"",
+ childname),
+ errdetail("The partition is being detached concurrently or has an unfinished detach."),
+ errhint("Use ALTER TABLE ... DETACH PARTITION ... FINALIZE to complete the pending detach operation")));
+ if (!inhdetached && detached)
+ ereport(ERROR,
+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("cannot complete detaching partition \"%s\"",
+ childname),
+ errdetail("There's no partial concurrent detach in progress.")));
+
CatalogTupleDelete(catalogRelation, &inheritsTuple->t_self);
found = true;
}
diff --git a/src/backend/commands/indexcmds.c b/src/backend/commands/indexcmds.c
index 2baca12c5f..d032fe4af3 100644
--- a/src/backend/commands/indexcmds.c
+++ b/src/backend/commands/indexcmds.c
@@ -1092,7 +1092,7 @@ DefineIndex(Oid relationId,
*/
if (partitioned && stmt->relation && !stmt->relation->inh)
{
- PartitionDesc pd = RelationGetPartitionDesc(rel);
+ PartitionDesc pd = RelationGetPartitionDesc(rel, false);
if (pd->nparts != 0)
flags |= INDEX_CREATE_INVALID;
@@ -1149,7 +1149,7 @@ DefineIndex(Oid relationId,
*/
if (!stmt->relation || stmt->relation->inh)
{
- PartitionDesc partdesc = RelationGetPartitionDesc(rel);
+ PartitionDesc partdesc = RelationGetPartitionDesc(rel, false);
int nparts = partdesc->nparts;
Oid *part_oids = palloc(sizeof(Oid) * nparts);
bool invalidate_parent = false;
@@ -3550,6 +3550,7 @@ IndexSetParentIndex(Relation partitionIdx, Oid parentOid)
values[Anum_pg_inherits_inhparent - 1] =
ObjectIdGetDatum(parentOid);
values[Anum_pg_inherits_inhseqno - 1] = Int32GetDatum(1);
+ values[Anum_pg_inherits_inhdetached - 1] = BoolGetDatum(false);
memset(isnull, false, sizeof(isnull));
tuple = heap_form_tuple(RelationGetDescr(pg_inherits),
diff --git a/src/backend/commands/tablecmds.c b/src/backend/commands/tablecmds.c
index 07f4f562ee..682aa6a015 100644
--- a/src/backend/commands/tablecmds.c
+++ b/src/backend/commands/tablecmds.c
@@ -541,7 +541,8 @@ static PartitionSpec *transformPartitionSpec(Relation rel, PartitionSpec *partsp
static void ComputePartitionAttrs(ParseState *pstate, Relation rel, List *partParams, AttrNumber *partattrs,
List **partexprs, Oid *partopclass, Oid *partcollation, char strategy);
static void CreateInheritance(Relation child_rel, Relation parent_rel);
-static void RemoveInheritance(Relation child_rel, Relation parent_rel);
+static void RemoveInheritance(Relation child_rel, Relation parent_rel,
+ bool allow_detached);
static ObjectAddress ATExecAttachPartition(List **wqueue, Relation rel,
PartitionCmd *cmd);
static void AttachPartitionEnsureIndexes(Relation rel, Relation attachrel);
@@ -550,7 +551,12 @@ static void QueuePartitionConstraintValidation(List **wqueue, Relation scanrel,
bool validate_default);
static void CloneRowTriggersToPartition(Relation parent, Relation partition);
static void DropClonedTriggersFromPartition(Oid partitionId);
-static ObjectAddress ATExecDetachPartition(Relation rel, RangeVar *name);
+static ObjectAddress ATExecDetachPartition(List **wqueue, AlteredTableInfo *tab,
+ Relation rel, RangeVar *name,
+ bool concurrent);
+static void DetachPartitionFinalize(Relation rel, Relation partRel,
+ bool concurrent, Oid defaultPartOid);
+static ObjectAddress ATExecDetachPartitionFinalize(Relation rel, RangeVar *name);
static ObjectAddress ATExecAttachPartitionIdx(List **wqueue, Relation rel,
RangeVar *name);
static void validatePartitionedIndex(Relation partedIdx, Relation partedTbl);
@@ -984,7 +990,8 @@ DefineRelation(CreateStmt *stmt, char relkind, Oid ownerId,
* lock the partition so as to avoid a deadlock.
*/
defaultPartOid =
- get_default_oid_from_partdesc(RelationGetPartitionDesc(parent));
+ get_default_oid_from_partdesc(RelationGetPartitionDesc(parent,
+ false));
if (OidIsValid(defaultPartOid))
defaultRel = table_open(defaultPartOid, AccessExclusiveLock);
@@ -3099,7 +3106,7 @@ renameatt_internal(Oid myrelid,
* expected_parents will only be 0 if we are not already recursing.
*/
if (expected_parents == 0 &&
- find_inheritance_children(myrelid, NoLock) != NIL)
+ find_inheritance_children(myrelid, false, NoLock) != NIL)
ereport(ERROR,
(errcode(ERRCODE_INVALID_TABLE_DEFINITION),
errmsg("inherited column \"%s\" must be renamed in child tables too",
@@ -3298,7 +3305,7 @@ rename_constraint_internal(Oid myrelid,
else
{
if (expected_parents == 0 &&
- find_inheritance_children(myrelid, NoLock) != NIL)
+ find_inheritance_children(myrelid, false, NoLock) != NIL)
ereport(ERROR,
(errcode(ERRCODE_INVALID_TABLE_DEFINITION),
errmsg("inherited constraint \"%s\" must be renamed in child tables too",
@@ -3916,7 +3923,14 @@ AlterTableGetLockLevel(List *cmds)
break;
case AT_DetachPartition:
- cmd_lockmode = AccessExclusiveLock;
+ if (((PartitionCmd *) cmd->def)->concurrent)
+ cmd_lockmode = ShareUpdateExclusiveLock;
+ else
+ cmd_lockmode = AccessExclusiveLock;
+ break;
+
+ case AT_DetachPartitionFinalize:
+ cmd_lockmode = ShareUpdateExclusiveLock;
break;
case AT_CheckNotNull:
@@ -4288,6 +4302,11 @@ ATPrepCmd(List **wqueue, Relation rel, AlterTableCmd *cmd,
/* No command-specific prep needed */
pass = AT_PASS_MISC;
break;
+ case AT_DetachPartitionFinalize:
+ ATSimplePermissions(rel, ATT_TABLE);
+ /* No command-specific prep needed */
+ pass = AT_PASS_MISC;
+ break;
default: /* oops */
elog(ERROR, "unrecognized alter table type: %d",
(int) cmd->subtype);
@@ -4663,7 +4682,12 @@ ATExecCmd(List **wqueue, AlteredTableInfo *tab,
Assert(cmd != NULL);
/* ATPrepCmd ensures it must be a table */
Assert(rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE);
- ATExecDetachPartition(rel, ((PartitionCmd *) cmd->def)->name);
+ ATExecDetachPartition(wqueue, tab, rel,
+ ((PartitionCmd *) cmd->def)->name,
+ ((PartitionCmd *) cmd->def)->concurrent);
+ break;
+ case AT_DetachPartitionFinalize:
+ ATExecDetachPartitionFinalize(rel, ((PartitionCmd *) cmd->def)->name);
break;
default: /* oops */
elog(ERROR, "unrecognized alter table type: %d",
@@ -6081,7 +6105,7 @@ ATExecAddColumn(List **wqueue, AlteredTableInfo *tab, Relation rel,
*/
if (colDef->identity &&
recurse &&
- find_inheritance_children(myrelid, NoLock) != NIL)
+ find_inheritance_children(myrelid, false, NoLock) != NIL)
ereport(ERROR,
(errcode(ERRCODE_INVALID_TABLE_DEFINITION),
errmsg("cannot recursively add identity column to table that has child tables")));
@@ -6314,7 +6338,8 @@ ATExecAddColumn(List **wqueue, AlteredTableInfo *tab, Relation rel,
* routines, we have to do this one level of recursion at a time; we can't
* use find_all_inheritors to do it in one pass.
*/
- children = find_inheritance_children(RelationGetRelid(rel), lockmode);
+ children =
+ find_inheritance_children(RelationGetRelid(rel), false, lockmode);
/*
* If we are told not to recurse, there had better not be any child
@@ -6468,7 +6493,7 @@ ATPrepDropNotNull(Relation rel, bool recurse, bool recursing)
*/
if (rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE)
{
- PartitionDesc partdesc = RelationGetPartitionDesc(rel);
+ PartitionDesc partdesc = RelationGetPartitionDesc(rel, false);
Assert(partdesc != NULL);
if (partdesc->nparts > 0 && !recurse && !recursing)
@@ -7680,7 +7705,8 @@ ATExecDropColumn(List **wqueue, Relation rel, const char *colName,
* routines, we have to do this one level of recursion at a time; we can't
* use find_all_inheritors to do it in one pass.
*/
- children = find_inheritance_children(RelationGetRelid(rel), lockmode);
+ children =
+ find_inheritance_children(RelationGetRelid(rel), false, lockmode);
if (children)
{
@@ -8144,7 +8170,8 @@ ATAddCheckConstraint(List **wqueue, AlteredTableInfo *tab, Relation rel,
* routines, we have to do this one level of recursion at a time; we can't
* use find_all_inheritors to do it in one pass.
*/
- children = find_inheritance_children(RelationGetRelid(rel), lockmode);
+ children =
+ find_inheritance_children(RelationGetRelid(rel), false, lockmode);
/*
* Check if ONLY was specified with ALTER TABLE. If so, allow the
@@ -8759,7 +8786,7 @@ addFkRecurseReferenced(List **wqueue, Constraint *fkconstraint, Relation rel,
*/
if (pkrel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE)
{
- PartitionDesc pd = RelationGetPartitionDesc(pkrel);
+ PartitionDesc pd = RelationGetPartitionDesc(pkrel, false);
for (int i = 0; i < pd->nparts; i++)
{
@@ -8893,7 +8920,7 @@ addFkRecurseReferencing(List **wqueue, Constraint *fkconstraint, Relation rel,
}
else if (rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE)
{
- PartitionDesc pd = RelationGetPartitionDesc(rel);
+ PartitionDesc pd = RelationGetPartitionDesc(rel, false);
/*
* Recurse to take appropriate action on each partition; either we
@@ -10674,7 +10701,8 @@ ATExecDropConstraint(Relation rel, const char *constrName,
* use find_all_inheritors to do it in one pass.
*/
if (!is_no_inherit_constraint)
- children = find_inheritance_children(RelationGetRelid(rel), lockmode);
+ children =
+ find_inheritance_children(RelationGetRelid(rel), false, lockmode);
else
children = NIL;
@@ -11058,7 +11086,8 @@ ATPrepAlterColumnType(List **wqueue,
}
}
else if (!recursing &&
- find_inheritance_children(RelationGetRelid(rel), NoLock) != NIL)
+ find_inheritance_children(RelationGetRelid(rel), false,
+ NoLock) != NIL)
ereport(ERROR,
(errcode(ERRCODE_INVALID_TABLE_DEFINITION),
errmsg("type of inherited column \"%s\" must be changed in child tables too",
@@ -13953,7 +13982,7 @@ ATExecDropInherit(Relation rel, RangeVar *parent, LOCKMODE lockmode)
*/
/* Off to RemoveInheritance() where most of the work happens */
- RemoveInheritance(rel, parent_rel);
+ RemoveInheritance(rel, parent_rel, false);
ObjectAddressSet(address, RelationRelationId,
RelationGetRelid(parent_rel));
@@ -13964,12 +13993,70 @@ ATExecDropInherit(Relation rel, RangeVar *parent, LOCKMODE lockmode)
return address;
}
+/*
+ * MarkInheritDetached
+ *
+ * When a partition is detached from its parent concurrently, we don't
+ * remove the pg_inherits row until a second transaction; as a preparatory
+ * step, this function marks the entry as 'detached', so that other
+ * transactions know to ignore the partition.
+ */
+static void
+MarkInheritDetached(Relation child_rel, Relation parent_rel)
+{
+ Relation catalogRelation;
+ SysScanDesc scan;
+ ScanKeyData key;
+ HeapTuple inheritsTuple;
+ bool found = false;
+
+ /*
+ * Find pg_inherits entries by inhrelid.
+ */
+ catalogRelation = table_open(InheritsRelationId, RowExclusiveLock);
+ ScanKeyInit(&key,
+ Anum_pg_inherits_inhrelid,
+ BTEqualStrategyNumber, F_OIDEQ,
+ ObjectIdGetDatum(RelationGetRelid(child_rel)));
+ scan = systable_beginscan(catalogRelation, InheritsRelidSeqnoIndexId,
+ true, NULL, 1, &key);
+
+ while (HeapTupleIsValid(inheritsTuple = systable_getnext(scan)))
+ {
+ HeapTuple newtup;
+
+ if (((Form_pg_inherits) GETSTRUCT(inheritsTuple))->inhparent !=
+ RelationGetRelid(parent_rel))
+ elog(ERROR, "bad parent tuple found for partition %u",
+ RelationGetRelid(child_rel));
+
+ newtup = heap_copytuple(inheritsTuple);
+ ((Form_pg_inherits) GETSTRUCT(newtup))->inhdetached = true;
+
+ CatalogTupleUpdate(catalogRelation,
+ &inheritsTuple->t_self,
+ newtup);
+ found = true;
+ }
+
+ /* Done */
+ systable_endscan(scan);
+ table_close(catalogRelation, RowExclusiveLock);
+
+ if (!found)
+ ereport(ERROR,
+ (errcode(ERRCODE_UNDEFINED_TABLE),
+ errmsg("relation \"%s\" is not a partition of relation \"%s\"",
+ RelationGetRelationName(child_rel),
+ RelationGetRelationName(parent_rel))));
+}
+
/*
* RemoveInheritance
*
* Drop a parent from the child's parents. This just adjusts the attinhcount
* and attislocal of the columns and removes the pg_inherit and pg_depend
- * entries.
+ * entries. expect_detached is passed down to DeleteInheritsTuple, q.v..
*
* If attinhcount goes to 0 then attislocal gets set to true. If it goes back
* up attislocal stays true, which means if a child is ever removed from a
@@ -13983,7 +14070,7 @@ ATExecDropInherit(Relation rel, RangeVar *parent, LOCKMODE lockmode)
* Common to ATExecDropInherit() and ATExecDetachPartition().
*/
static void
-RemoveInheritance(Relation child_rel, Relation parent_rel)
+RemoveInheritance(Relation child_rel, Relation parent_rel, bool expect_detached)
{
Relation catalogRelation;
SysScanDesc scan;
@@ -13999,7 +14086,9 @@ RemoveInheritance(Relation child_rel, Relation parent_rel)
child_is_partition = true;
found = DeleteInheritsTuple(RelationGetRelid(child_rel),
- RelationGetRelid(parent_rel));
+ RelationGetRelid(parent_rel),
+ expect_detached,
+ RelationGetRelationName(child_rel));
if (!found)
{
if (child_is_partition)
@@ -16107,7 +16196,7 @@ QueuePartitionConstraintValidation(List **wqueue, Relation scanrel,
}
else if (scanrel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE)
{
- PartitionDesc partdesc = RelationGetPartitionDesc(scanrel);
+ PartitionDesc partdesc = RelationGetPartitionDesc(scanrel, false);
int i;
for (i = 0; i < partdesc->nparts; i++)
@@ -16163,7 +16252,7 @@ ATExecAttachPartition(List **wqueue, Relation rel, PartitionCmd *cmd)
* new partition will change its partition constraint.
*/
defaultPartOid =
- get_default_oid_from_partdesc(RelationGetPartitionDesc(rel));
+ get_default_oid_from_partdesc(RelationGetPartitionDesc(rel, false));
if (OidIsValid(defaultPartOid))
LockRelationOid(defaultPartOid, AccessExclusiveLock);
@@ -16751,105 +16840,226 @@ CloneRowTriggersToPartition(Relation parent, Relation partition)
* ALTER TABLE DETACH PARTITION
*
* Return the address of the relation that is no longer a partition of rel.
+ *
+ * If concurrent mode is requested, we run in two transactions. A side-
+ * effect is that this command cannot run in a multi-part ALTER TABLE.
+ * Currently, that's enforced by the grammar.
+ *
+ * The strategy for concurrency is to first modify the partition catalog
+ * rows to make it visible to everyone that the partition is detached,
+ * lock the partition against writes, and commit the transaction; anyone
+ * who requests the partition descriptor from that point onwards has to
+ * ignore such a partition. In a second transaction, we wait until all
+ * transactions that could have seen the partition as attached are gone,
+ * then we remove the rest of partition metadata (pg_inherits and
+ * pg_class.relpartbounds).
*/
static ObjectAddress
-ATExecDetachPartition(Relation rel, RangeVar *name)
+ATExecDetachPartition(List **wqueue, AlteredTableInfo *tab, Relation rel,
+ RangeVar *name, bool concurrent)
{
- Relation partRel,
- classRel;
- HeapTuple tuple,
- newtuple;
- Datum new_val[Natts_pg_class];
- bool new_null[Natts_pg_class],
- new_repl[Natts_pg_class];
+ Relation partRel;
ObjectAddress address;
Oid defaultPartOid;
- List *indexes;
- List *fks;
- ListCell *cell;
/*
* We must lock the default partition, because detaching this partition
* will change its partition constraint.
*/
defaultPartOid =
- get_default_oid_from_partdesc(RelationGetPartitionDesc(rel));
- if (OidIsValid(defaultPartOid))
- LockRelationOid(defaultPartOid, AccessExclusiveLock);
-
- partRel = table_openrv(name, ShareUpdateExclusiveLock);
-
- /* Ensure that foreign keys still hold after this detach */
- ATDetachCheckNoForeignKeyRefs(partRel);
-
- /* All inheritance related checks are performed within the function */
- RemoveInheritance(partRel, rel);
-
- /* Update pg_class tuple */
- classRel = table_open(RelationRelationId, RowExclusiveLock);
- tuple = SearchSysCacheCopy1(RELOID,
- ObjectIdGetDatum(RelationGetRelid(partRel)));
- if (!HeapTupleIsValid(tuple))
- elog(ERROR, "cache lookup failed for relation %u",
- RelationGetRelid(partRel));
- Assert(((Form_pg_class) GETSTRUCT(tuple))->relispartition);
-
- /* Clear relpartbound and reset relispartition */
- memset(new_val, 0, sizeof(new_val));
- memset(new_null, false, sizeof(new_null));
- memset(new_repl, false, sizeof(new_repl));
- new_val[Anum_pg_class_relpartbound - 1] = (Datum) 0;
- new_null[Anum_pg_class_relpartbound - 1] = true;
- new_repl[Anum_pg_class_relpartbound - 1] = true;
- newtuple = heap_modify_tuple(tuple, RelationGetDescr(classRel),
- new_val, new_null, new_repl);
-
- ((Form_pg_class) GETSTRUCT(newtuple))->relispartition = false;
- CatalogTupleUpdate(classRel, &newtuple->t_self, newtuple);
- heap_freetuple(newtuple);
-
+ get_default_oid_from_partdesc(RelationGetPartitionDesc(rel, false));
if (OidIsValid(defaultPartOid))
{
/*
- * If the relation being detached is the default partition itself,
- * remove it from the parent's pg_partitioned_table entry.
+ * Concurrent detaching when a default partition exists is not
+ * supported. The main problem is that the default partition
+ * constraint would change. And there's a definitional problem: what
+ * should happen to the tuples that are being inserted that belong to
+ * the partition being detached? Putting them on the partition being
+ * detached would be wrong, since they'd become "lost" after the but
+ * we cannot put them in the default partition either until we alter
+ * its partition constraint.
*
- * If not, we must invalidate default partition's relcache entry, as
- * in StorePartitionBound: its partition constraint depends on every
- * other partition's partition constraint.
+ * I think we could solve this problem if we effected the constraint
+ * change before committing the first transaction. But the lock would
+ * have to remain AEL and it would cause concurrent query planning to
+ * be blocked, so changing it that way would be even worse.
*/
- if (RelationGetRelid(partRel) == defaultPartOid)
- update_default_partition_oid(RelationGetRelid(rel), InvalidOid);
- else
- CacheInvalidateRelcacheByRelid(defaultPartOid);
+ if (concurrent)
+ ereport(ERROR,
+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("cannot detach partitions concurrently when a default partition exists")));
+ LockRelationOid(defaultPartOid, AccessExclusiveLock);
}
- /* detach indexes too */
- indexes = RelationGetIndexList(partRel);
- foreach(cell, indexes)
+ /*
+ * In concurrent mode, the partition is locked with
+ * ShareUpdateExclusiveLock in the first stage.
+ */
+ partRel = table_openrv(name, concurrent ?
+ ShareUpdateExclusiveLock : AccessExclusiveLock);
+
+ /*
+ * Check inheritance conditions and either delete the pg_inherits row
+ * (in non-concurrent mode) or just set the inhisdetached flag.
+ */
+ if (!concurrent)
+ RemoveInheritance(partRel, rel, false);
+ else
+ MarkInheritDetached(partRel, rel);
+
+ /*
+ * Ensure that foreign keys still hold after this detach. This keeps lock
+ * on the referencing tables, which prevent concurrent transactions from
+ * adding rows that we wouldn't see. For this to work in concurrent mode,
+ * it is critical that the partition appears as no longer attached for the
+ * RI queries as soon as the first transaction commits.
+ */
+ ATDetachCheckNoForeignKeyRefs(partRel);
+
+ /*
+ * Concurrent mode has to work harder; first we add a new constraint to the
+ * partition that matches the partition constraint. The reason for this is
+ * that the planner may have made optimizations that depend on the
+ * constraint. XXX Isn't it sufficient to invalidate the partition's
+ * relcache entry?
+ *
+ * Then we close our existing transaction, and in a new one wait for
+ * all process to catch up on the catalog updates we've done so far; at
+ * that point we can complete the operation.
+ */
+ if (concurrent)
{
- Oid idxid = lfirst_oid(cell);
- Relation idx;
- Oid constrOid;
+ Oid partrelid,
+ parentrelid;
+ Constraint *n;
+ LOCKTAG tag;
+ char *parentrelname;
+ char *partrelname;
- if (!has_superclass(idxid))
- continue;
+ /* Add constraint on partition, equivalent to the partition constraint */
+ n = makeNode(Constraint);
+ n->contype = CONSTR_CHECK;
+ n->conname = NULL;
+ n->location = -1;
+ n->is_no_inherit = false;
+ n->raw_expr = NULL;
+ n->cooked_expr = nodeToString(make_ands_explicit(RelationGetPartitionQual(partRel)));
+ n->initially_valid = true;
+ n->skip_validation = true;
+ /* It's a re-add, since it nominally already exists */
+ ATAddCheckConstraint(wqueue, tab, partRel, n,
+ true, false, true, ShareUpdateExclusiveLock);
- Assert((IndexGetRelation(get_partition_parent(idxid), false) ==
- RelationGetRelid(rel)));
+ /*
+ * We're almost done now; the only traces that remain are the
+ * pg_inherits tuple and the partition's relpartbounds. Before we can
+ * remove those, we need to wait until all transactions that know that
+ * this is a partition are gone.
+ */
- idx = index_open(idxid, AccessExclusiveLock);
- IndexSetParentIndex(idx, InvalidOid);
+ /*
+ * Remember relation OIDs to re-acquire them later; and relation names
+ * too, for error messages if something is dropped in between.
+ */
+ partrelid = RelationGetRelid(partRel);
+ parentrelid = RelationGetRelid(rel);
+ parentrelname = MemoryContextStrdup(PortalContext,
+ RelationGetRelationName(rel));
+ partrelname = MemoryContextStrdup(PortalContext,
+ RelationGetRelationName(partRel));
- /* If there's a constraint associated with the index, detach it too */
- constrOid = get_relation_idx_constraint_oid(RelationGetRelid(partRel),
- idxid);
- if (OidIsValid(constrOid))
- ConstraintSetParentConstraint(constrOid, InvalidOid, InvalidOid);
+ /* Invalidate relcache entries for the parent -- must be before close */
+ CacheInvalidateRelcache(rel);
+
+ table_close(partRel, NoLock);
+ table_close(rel, NoLock);
+ tab->rel = NULL;
+
+ /* Make updated catalog entry visible */
+ PopActiveSnapshot();
+ CommitTransactionCommand();
+
+ StartTransactionCommand();
+
+ /*
+ * Now wait. This ensures that all queries that were planned including
+ * the partition are finished before we remove the rest of catalog
+ * entries. We don't need or indeed want to acquire this lock, though
+ * -- that would block later queries.
+ *
+ * We don't need to concern ourselves with waiting for a lock the
+ * partition itself, since we will acquire AccessExclusiveLock below.
+ */
+ SET_LOCKTAG_RELATION(tag, MyDatabaseId, parentrelid);
+ WaitForLockersMultiple(list_make1(&tag), AccessExclusiveLock, false);
+
+ /*
+ * Now acquire locks in both relations again. Note they may have been
+ * removed in the meantime, so care is required.
+ */
+ rel = try_relation_open(parentrelid, ShareUpdateExclusiveLock);
+ partRel = try_relation_open(partrelid, AccessExclusiveLock);
+
+ /* If the relations aren't there, something bad happened; bail out */
+ if (rel == NULL)
+ {
+ if (partRel != NULL) /* shouldn't happen */
+ elog(WARNING, "dangling partition \"%s\" remains, can't fix",
+ partrelname);
+ ereport(ERROR,
+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("partitioned table \"%s\" was removed concurrently",
+ parentrelname)));
+ }
+ if (partRel == NULL)
+ ereport(ERROR,
+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("partition \"%s\" was removed concurrently", partrelname)));
+
+ tab->rel = rel;
- index_close(idx, NoLock);
}
- table_close(classRel, RowExclusiveLock);
+
+ /* Do the final part of detaching */
+ DetachPartitionFinalize(rel, partRel, concurrent, defaultPartOid);
+
+ ObjectAddressSet(address, RelationRelationId, RelationGetRelid(partRel));
+
+ /* keep our lock until commit */
+ table_close(partRel, NoLock);
+
+ return address;
+}
+
+/*
+ * Second part of ALTER TABLE .. DETACH.
+ *
+ * This is separate so that it can be run independently when the second
+ * transaction of the concurrent algorithm fails (crash or abort).
+ */
+static void
+DetachPartitionFinalize(Relation rel, Relation partRel, bool concurrent,
+ Oid defaultPartOid)
+{
+ Relation classRel;
+ List *fks;
+ ListCell *cell;
+ List *indexes;
+ Datum new_val[Natts_pg_class];
+ bool new_null[Natts_pg_class],
+ new_repl[Natts_pg_class];
+ HeapTuple tuple,
+ newtuple;
+
+ if (concurrent)
+ {
+ /*
+ * We can remove the pg_inherits row now. (In the non-concurrent case,
+ * this was already done).
+ */
+ RemoveInheritance(partRel, rel, true);
+ }
/* Drop any triggers that were cloned on creation/attach. */
DropClonedTriggersFromPartition(RelationGetRelid(partRel));
@@ -16922,17 +17132,98 @@ ATExecDetachPartition(Relation rel, RangeVar *name)
ObjectAddressSet(constraint, ConstraintRelationId, constrOid);
performDeletion(&constraint, DROP_RESTRICT, 0);
}
- CommandCounterIncrement();
+
+ /* Now we can detach indexes */
+ indexes = RelationGetIndexList(partRel);
+ foreach(cell, indexes)
+ {
+ Oid idxid = lfirst_oid(cell);
+ Relation idx;
+ Oid constrOid;
+
+ if (!has_superclass(idxid))
+ continue;
+
+ Assert((IndexGetRelation(get_partition_parent(idxid), false) ==
+ RelationGetRelid(rel)));
+
+ idx = index_open(idxid, AccessExclusiveLock);
+ IndexSetParentIndex(idx, InvalidOid);
+
+ /* If there's a constraint associated with the index, detach it too */
+ constrOid = get_relation_idx_constraint_oid(RelationGetRelid(partRel),
+ idxid);
+ if (OidIsValid(constrOid))
+ ConstraintSetParentConstraint(constrOid, InvalidOid, InvalidOid);
+
+ index_close(idx, NoLock);
+ }
+
+ /* Update pg_class tuple */
+ classRel = table_open(RelationRelationId, RowExclusiveLock);
+ tuple = SearchSysCacheCopy1(RELOID,
+ ObjectIdGetDatum(RelationGetRelid(partRel)));
+ if (!HeapTupleIsValid(tuple))
+ elog(ERROR, "cache lookup failed for relation %u",
+ RelationGetRelid(partRel));
+ Assert(((Form_pg_class) GETSTRUCT(tuple))->relispartition);
+
+ /* Clear relpartbound and reset relispartition */
+ memset(new_val, 0, sizeof(new_val));
+ memset(new_null, false, sizeof(new_null));
+ memset(new_repl, false, sizeof(new_repl));
+ new_val[Anum_pg_class_relpartbound - 1] = (Datum) 0;
+ new_null[Anum_pg_class_relpartbound - 1] = true;
+ new_repl[Anum_pg_class_relpartbound - 1] = true;
+ newtuple = heap_modify_tuple(tuple, RelationGetDescr(classRel),
+ new_val, new_null, new_repl);
+
+ ((Form_pg_class) GETSTRUCT(newtuple))->relispartition = false;
+ CatalogTupleUpdate(classRel, &newtuple->t_self, newtuple);
+ heap_freetuple(newtuple);
+ table_close(classRel, RowExclusiveLock);
+
+ if (OidIsValid(defaultPartOid))
+ {
+ /*
+ * If the relation being detached is the default partition itself,
+ * remove it from the parent's pg_partitioned_table entry.
+ *
+ * If not, we must invalidate default partition's relcache entry, as
+ * in StorePartitionBound: its partition constraint depends on every
+ * other partition's partition constraint.
+ */
+ if (RelationGetRelid(partRel) == defaultPartOid)
+ update_default_partition_oid(RelationGetRelid(rel), InvalidOid);
+ else
+ CacheInvalidateRelcacheByRelid(defaultPartOid);
+ }
/*
* Invalidate the parent's relcache so that the partition is no longer
* included in its partition descriptor.
*/
CacheInvalidateRelcache(rel);
+}
+
+/*
+ * ALTER TABLE ... DETACH PARTITION ... FINALIZE
+ *
+ * To use when a DETACH PARTITION command previously did not run to
+ * completion; this completes the detaching process.
+ */
+static ObjectAddress
+ATExecDetachPartitionFinalize(Relation rel, RangeVar *name)
+{
+ Relation partRel;
+ ObjectAddress address;
+
+ partRel = table_openrv(name, AccessExclusiveLock);
+
+ DetachPartitionFinalize(rel, partRel, true, InvalidOid);
ObjectAddressSet(address, RelationRelationId, RelationGetRelid(partRel));
- /* keep our lock until commit */
table_close(partRel, NoLock);
return address;
@@ -17133,7 +17424,7 @@ ATExecAttachPartitionIdx(List **wqueue, Relation parentIdx, RangeVar *name)
RelationGetRelationName(partIdx))));
/* Make sure it indexes a partition of the other index's table */
- partDesc = RelationGetPartitionDesc(parentTbl);
+ partDesc = RelationGetPartitionDesc(parentTbl, false);
found = false;
for (i = 0; i < partDesc->nparts; i++)
{
@@ -17287,7 +17578,7 @@ validatePartitionedIndex(Relation partedIdx, Relation partedTbl)
* If we found as many inherited indexes as the partitioned table has
* partitions, we're good; update pg_index to set indisvalid.
*/
- if (tuples == RelationGetPartitionDesc(partedTbl)->nparts)
+ if (tuples == RelationGetPartitionDesc(partedTbl, false)->nparts)
{
Relation idxRel;
HeapTuple newtup;
diff --git a/src/backend/commands/trigger.c b/src/backend/commands/trigger.c
index 672fccff5b..ce65abd57e 100644
--- a/src/backend/commands/trigger.c
+++ b/src/backend/commands/trigger.c
@@ -1054,7 +1054,7 @@ CreateTrigger(CreateTrigStmt *stmt, const char *queryString,
*/
if (partition_recurse)
{
- PartitionDesc partdesc = RelationGetPartitionDesc(rel);
+ PartitionDesc partdesc = RelationGetPartitionDesc(rel, false);
List *idxs = NIL;
List *childTbls = NIL;
ListCell *l;
@@ -1076,7 +1076,8 @@ CreateTrigger(CreateTrigStmt *stmt, const char *queryString,
ListCell *l;
List *idxs = NIL;
- idxs = find_inheritance_children(indexOid, ShareRowExclusiveLock);
+ idxs = find_inheritance_children(indexOid, false,
+ ShareRowExclusiveLock);
foreach(l, idxs)
childTbls = lappend_oid(childTbls,
IndexGetRelation(lfirst_oid(l),
@@ -1537,7 +1538,7 @@ EnableDisableTrigger(Relation rel, const char *tgname,
if (rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE &&
(TRIGGER_FOR_ROW(oldtrig->tgtype)))
{
- PartitionDesc partdesc = RelationGetPartitionDesc(rel);
+ PartitionDesc partdesc = RelationGetPartitionDesc(rel, false);
int i;
for (i = 0; i < partdesc->nparts; i++)
diff --git a/src/backend/executor/execPartition.c b/src/backend/executor/execPartition.c
index fb6ce49056..ef4fd7bf90 100644
--- a/src/backend/executor/execPartition.c
+++ b/src/backend/executor/execPartition.c
@@ -507,6 +507,7 @@ ExecInitPartitionInfo(ModifyTableState *mtstate, EState *estate,
int partidx)
{
ModifyTable *node = (ModifyTable *) mtstate->ps.plan;
+ Oid partOid = dispatch->partdesc->oids[partidx];
Relation rootrel = rootResultRelInfo->ri_RelationDesc,
partrel;
Relation firstResultRel = mtstate->resultRelInfo[0].ri_RelationDesc;
@@ -517,7 +518,7 @@ ExecInitPartitionInfo(ModifyTableState *mtstate, EState *estate,
oldcxt = MemoryContextSwitchTo(proute->memcxt);
- partrel = table_open(dispatch->partdesc->oids[partidx], RowExclusiveLock);
+ partrel = table_open(partOid, RowExclusiveLock);
leaf_part_rri = makeNode(ResultRelInfo);
InitResultRelInfo(leaf_part_rri,
@@ -997,7 +998,7 @@ ExecInitPartitionDispatchInfo(EState *estate,
if (estate->es_partition_directory == NULL)
estate->es_partition_directory =
- CreatePartitionDirectory(estate->es_query_cxt);
+ CreatePartitionDirectory(estate->es_query_cxt, true);
oldcxt = MemoryContextSwitchTo(proute->memcxt);
@@ -1559,7 +1560,7 @@ ExecCreatePartitionPruneState(PlanState *planstate,
if (estate->es_partition_directory == NULL)
estate->es_partition_directory =
- CreatePartitionDirectory(estate->es_query_cxt);
+ CreatePartitionDirectory(estate->es_query_cxt, true);
n_part_hierarchies = list_length(partitionpruneinfo->prune_infos);
Assert(n_part_hierarchies > 0);
@@ -1625,9 +1626,12 @@ ExecCreatePartitionPruneState(PlanState *planstate,
partrel);
/*
- * Initialize the subplan_map and subpart_map. Since detaching a
- * partition requires AccessExclusiveLock, no partitions can have
- * disappeared, nor can the bounds for any partition have changed.
+ * Initialize the subplan_map and subpart_map.
+ *
+ * Because we request detached partitions to be included, and
+ * detaching waits for old transactions, it is safe to assume that
+ * no partitions have disappeared since this query was planned.
+ *
* However, new partitions may have been added.
*/
Assert(partdesc->nparts >= pinfo->nparts);
diff --git a/src/backend/nodes/copyfuncs.c b/src/backend/nodes/copyfuncs.c
index 89c409de66..f48350bc2f 100644
--- a/src/backend/nodes/copyfuncs.c
+++ b/src/backend/nodes/copyfuncs.c
@@ -4670,6 +4670,7 @@ _copyPartitionCmd(const PartitionCmd *from)
COPY_NODE_FIELD(name);
COPY_NODE_FIELD(bound);
+ COPY_SCALAR_FIELD(concurrent);
return newnode;
}
diff --git a/src/backend/nodes/equalfuncs.c b/src/backend/nodes/equalfuncs.c
index e3f33c40be..01ff684f90 100644
--- a/src/backend/nodes/equalfuncs.c
+++ b/src/backend/nodes/equalfuncs.c
@@ -2932,6 +2932,7 @@ _equalPartitionCmd(const PartitionCmd *a, const PartitionCmd *b)
{
COMPARE_NODE_FIELD(name);
COMPARE_NODE_FIELD(bound);
+ COMPARE_SCALAR_FIELD(concurrent);
return true;
}
diff --git a/src/backend/optimizer/util/plancat.c b/src/backend/optimizer/util/plancat.c
index 25545029d7..75fe783b41 100644
--- a/src/backend/optimizer/util/plancat.c
+++ b/src/backend/optimizer/util/plancat.c
@@ -2129,7 +2129,7 @@ set_relation_partition_info(PlannerInfo *root, RelOptInfo *rel,
/* Create the PartitionDirectory infrastructure if we didn't already */
if (root->glob->partition_directory == NULL)
root->glob->partition_directory =
- CreatePartitionDirectory(CurrentMemoryContext);
+ CreatePartitionDirectory(CurrentMemoryContext, false);
partdesc = PartitionDirectoryLookup(root->glob->partition_directory,
relation);
diff --git a/src/backend/parser/gram.y b/src/backend/parser/gram.y
index dbb47d4982..da0b233695 100644
--- a/src/backend/parser/gram.y
+++ b/src/backend/parser/gram.y
@@ -649,7 +649,7 @@ static Node *makeRecursiveViewSelect(char *relname, List *aliases, Node *query);
EXCLUDE EXCLUDING EXCLUSIVE EXECUTE EXISTS EXPLAIN EXPRESSION
EXTENSION EXTERNAL EXTRACT
- FALSE_P FAMILY FETCH FILTER FIRST_P FLOAT_P FOLLOWING FOR
+ FALSE_P FAMILY FETCH FILTER FINALIZE FIRST_P FLOAT_P FOLLOWING FOR
FORCE FOREIGN FORWARD FREEZE FROM FULL FUNCTION FUNCTIONS
GENERATED GLOBAL GRANT GRANTED GREATEST GROUP_P GROUPING GROUPS
@@ -2057,12 +2057,13 @@ partition_cmd:
n->subtype = AT_AttachPartition;
cmd->name = $3;
cmd->bound = $4;
+ cmd->concurrent = false;
n->def = (Node *) cmd;
$$ = (Node *) n;
}
- /* ALTER TABLE <name> DETACH PARTITION <partition_name> */
- | DETACH PARTITION qualified_name
+ /* ALTER TABLE <name> DETACH PARTITION <partition_name> [CONCURRENTLY] */
+ | DETACH PARTITION qualified_name opt_concurrently
{
AlterTableCmd *n = makeNode(AlterTableCmd);
PartitionCmd *cmd = makeNode(PartitionCmd);
@@ -2070,8 +2071,21 @@ partition_cmd:
n->subtype = AT_DetachPartition;
cmd->name = $3;
cmd->bound = NULL;
+ cmd->concurrent = $4;
n->def = (Node *) cmd;
+ $$ = (Node *) n;
+ }
+ | DETACH PARTITION qualified_name FINALIZE
+ {
+ AlterTableCmd *n = makeNode(AlterTableCmd);
+ PartitionCmd *cmd = makeNode(PartitionCmd);
+
+ n->subtype = AT_DetachPartitionFinalize;
+ cmd->name = $3;
+ cmd->bound = NULL;
+ cmd->concurrent = false;
+ n->def = (Node *) cmd;
$$ = (Node *) n;
}
;
@@ -2086,6 +2100,7 @@ index_partition_cmd:
n->subtype = AT_AttachPartition;
cmd->name = $3;
cmd->bound = NULL;
+ cmd->concurrent = false;
n->def = (Node *) cmd;
$$ = (Node *) n;
@@ -15110,6 +15125,7 @@ unreserved_keyword:
| EXTERNAL
| FAMILY
| FILTER
+ | FINALIZE
| FIRST_P
| FOLLOWING
| FORCE
diff --git a/src/backend/partitioning/partbounds.c b/src/backend/partitioning/partbounds.c
index 7553d55987..966d4bbe7f 100644
--- a/src/backend/partitioning/partbounds.c
+++ b/src/backend/partitioning/partbounds.c
@@ -2810,7 +2810,7 @@ check_new_partition_bound(char *relname, Relation parent,
PartitionBoundSpec *spec)
{
PartitionKey key = RelationGetPartitionKey(parent);
- PartitionDesc partdesc = RelationGetPartitionDesc(parent);
+ PartitionDesc partdesc = RelationGetPartitionDesc(parent, true);
PartitionBoundInfo boundinfo = partdesc->boundinfo;
ParseState *pstate = make_parsestate(NULL);
int with = -1;
@@ -3984,7 +3984,7 @@ get_qual_for_list(Relation parent, PartitionBoundSpec *spec)
{
int i;
int ndatums = 0;
- PartitionDesc pdesc = RelationGetPartitionDesc(parent);
+ PartitionDesc pdesc = RelationGetPartitionDesc(parent, true); /* XXX correct? */
PartitionBoundInfo boundinfo = pdesc->boundinfo;
if (boundinfo)
@@ -4184,7 +4184,7 @@ get_qual_for_range(Relation parent, PartitionBoundSpec *spec,
if (spec->is_default)
{
List *or_expr_args = NIL;
- PartitionDesc pdesc = RelationGetPartitionDesc(parent);
+ PartitionDesc pdesc = RelationGetPartitionDesc(parent, true); /* XXX correct? */
Oid *inhoids = pdesc->oids;
int nparts = pdesc->nparts,
i;
diff --git a/src/backend/partitioning/partdesc.c b/src/backend/partitioning/partdesc.c
index 0f124a5254..3c21f42ea9 100644
--- a/src/backend/partitioning/partdesc.c
+++ b/src/backend/partitioning/partdesc.c
@@ -38,6 +38,7 @@ typedef struct PartitionDirectoryData
{
MemoryContext pdir_mcxt;
HTAB *pdir_hash;
+ bool include_detached;
} PartitionDirectoryData;
typedef struct PartitionDirectoryEntry
@@ -47,7 +48,7 @@ typedef struct PartitionDirectoryEntry
PartitionDesc pd;
} PartitionDirectoryEntry;
-static void RelationBuildPartitionDesc(Relation rel);
+static void RelationBuildPartitionDesc(Relation rel, bool include_detached);
/*
@@ -62,13 +63,14 @@ static void RelationBuildPartitionDesc(Relation rel);
* that the data doesn't become stale.
*/
PartitionDesc
-RelationGetPartitionDesc(Relation rel)
+RelationGetPartitionDesc(Relation rel, bool include_detached)
{
if (rel->rd_rel->relkind != RELKIND_PARTITIONED_TABLE)
return NULL;
- if (unlikely(rel->rd_partdesc == NULL))
- RelationBuildPartitionDesc(rel);
+ if (unlikely(rel->rd_partdesc == NULL ||
+ rel->rd_partdesc->includes_detached != include_detached))
+ RelationBuildPartitionDesc(rel, include_detached);
return rel->rd_partdesc;
}
@@ -89,7 +91,7 @@ RelationGetPartitionDesc(Relation rel)
* permanently.
*/
static void
-RelationBuildPartitionDesc(Relation rel)
+RelationBuildPartitionDesc(Relation rel, bool include_detached)
{
PartitionDesc partdesc;
PartitionBoundInfo boundinfo = NULL;
@@ -111,7 +113,8 @@ RelationBuildPartitionDesc(Relation rel)
* concurrently, whatever this function returns will be accurate as of
* some well-defined point in time.
*/
- inhoids = find_inheritance_children(RelationGetRelid(rel), NoLock);
+ inhoids = find_inheritance_children(RelationGetRelid(rel), NoLock,
+ include_detached);
nparts = list_length(inhoids);
/* Allocate working arrays for OIDs, leaf flags, and boundspecs. */
@@ -239,6 +242,7 @@ RelationBuildPartitionDesc(Relation rel)
partdesc->boundinfo = partition_bounds_copy(boundinfo, key);
partdesc->oids = (Oid *) palloc(nparts * sizeof(Oid));
partdesc->is_leaf = (bool *) palloc(nparts * sizeof(bool));
+ partdesc->includes_detached = include_detached;
/*
* Assign OIDs from the original array into mapped indexes of the
@@ -281,7 +285,7 @@ RelationBuildPartitionDesc(Relation rel)
* Create a new partition directory object.
*/
PartitionDirectory
-CreatePartitionDirectory(MemoryContext mcxt)
+CreatePartitionDirectory(MemoryContext mcxt, bool include_detached)
{
MemoryContext oldcontext = MemoryContextSwitchTo(mcxt);
PartitionDirectory pdir;
@@ -296,6 +300,7 @@ CreatePartitionDirectory(MemoryContext mcxt)
pdir->pdir_mcxt = mcxt;
pdir->pdir_hash = hash_create("partition directory", 256, &ctl,
HASH_ELEM | HASH_BLOBS | HASH_CONTEXT);
+ pdir->include_detached = include_detached;
MemoryContextSwitchTo(oldcontext);
return pdir;
@@ -328,7 +333,7 @@ PartitionDirectoryLookup(PartitionDirectory pdir, Relation rel)
*/
RelationIncrementReferenceCount(rel);
pde->rel = rel;
- pde->pd = RelationGetPartitionDesc(rel);
+ pde->pd = RelationGetPartitionDesc(rel, pdir->include_detached);
Assert(pde->pd != NULL);
}
return pde->pd;
diff --git a/src/backend/tcop/utility.c b/src/backend/tcop/utility.c
index 9b0c376c8c..57ce760edc 100644
--- a/src/backend/tcop/utility.c
+++ b/src/backend/tcop/utility.c
@@ -1240,6 +1240,25 @@ ProcessUtilitySlow(ParseState *pstate,
AlterTableStmt *atstmt = (AlterTableStmt *) parsetree;
Oid relid;
LOCKMODE lockmode;
+ ListCell *cell;
+
+ /*
+ * Disallow ALTER TABLE .. DETACH CONCURRENTLY in a
+ * transaction block or function. (Perhaps it could be
+ * allowed in a procedure, but don't hold your breath.)
+ */
+ foreach(cell, atstmt->cmds)
+ {
+ AlterTableCmd *cmd = (AlterTableCmd *) lfirst(cell);
+
+ /* Disallow DETACH CONCURRENTLY in a transaction block */
+ if (cmd->subtype == AT_DetachPartition)
+ {
+ if (((PartitionCmd *) cmd->def)->concurrent)
+ PreventInTransactionBlock(isTopLevel,
+ "ALTER TABLE ... DETACH CONCURRENTLY");
+ }
+ }
/*
* Figure out lock mode, and acquire lock. This also does
diff --git a/src/bin/psql/describe.c b/src/bin/psql/describe.c
index 57266f4fc3..329b650e12 100644
--- a/src/bin/psql/describe.c
+++ b/src/bin/psql/describe.c
@@ -2109,7 +2109,12 @@ describeOneTableDetails(const char *schemaname,
printfPQExpBuffer(&buf,
"SELECT inhparent::pg_catalog.regclass,\n"
- " pg_catalog.pg_get_expr(c.relpartbound, c.oid)");
+ " pg_catalog.pg_get_expr(c.relpartbound, c.oid),\n ");
+
+ appendPQExpBuffer(&buf,
+ pset.sversion >= 140000 ? "inhdetached" :
+ "false as inhdetached");
+
/* If verbose, also request the partition constraint definition */
if (verbose)
appendPQExpBufferStr(&buf,
@@ -2127,17 +2132,19 @@ describeOneTableDetails(const char *schemaname,
{
char *parent_name = PQgetvalue(result, 0, 0);
char *partdef = PQgetvalue(result, 0, 1);
+ char *detached = PQgetvalue(result, 0, 2);
- printfPQExpBuffer(&tmpbuf, _("Partition of: %s %s"), parent_name,
- partdef);
+ printfPQExpBuffer(&tmpbuf, _("Partition of: %s %s%s"), parent_name,
+ partdef,
+ strcmp(detached, "t") == 0 ? " DETACHED" : "");
printTableAddFooter(&cont, tmpbuf.data);
if (verbose)
{
char *partconstraintdef = NULL;
- if (!PQgetisnull(result, 0, 2))
- partconstraintdef = PQgetvalue(result, 0, 2);
+ if (!PQgetisnull(result, 0, 3))
+ partconstraintdef = PQgetvalue(result, 0, 3);
/* If there isn't any constraint, show that explicitly */
if (partconstraintdef == NULL || partconstraintdef[0] == '\0')
printfPQExpBuffer(&tmpbuf, _("No partition constraint"));
@@ -3179,9 +3186,18 @@ describeOneTableDetails(const char *schemaname,
}
/* print child tables (with additional info if partitions) */
- if (pset.sversion >= 100000)
+ if (pset.sversion >= 140000)
printfPQExpBuffer(&buf,
- "SELECT c.oid::pg_catalog.regclass, c.relkind,"
+ "SELECT c.oid::pg_catalog.regclass, c.relkind, inhdetached,"
+ " pg_catalog.pg_get_expr(c.relpartbound, c.oid)\n"
+ "FROM pg_catalog.pg_class c, pg_catalog.pg_inherits i\n"
+ "WHERE c.oid = i.inhrelid AND i.inhparent = '%s'\n"
+ "ORDER BY pg_catalog.pg_get_expr(c.relpartbound, c.oid) = 'DEFAULT',"
+ " c.oid::pg_catalog.regclass::pg_catalog.text;",
+ oid);
+ else if (pset.sversion >= 100000)
+ printfPQExpBuffer(&buf,
+ "SELECT c.oid::pg_catalog.regclass, c.relkind, false AS inhdetached,"
" pg_catalog.pg_get_expr(c.relpartbound, c.oid)\n"
"FROM pg_catalog.pg_class c, pg_catalog.pg_inherits i\n"
"WHERE c.oid = i.inhrelid AND i.inhparent = '%s'\n"
@@ -3190,14 +3206,14 @@ describeOneTableDetails(const char *schemaname,
oid);
else if (pset.sversion >= 80300)
printfPQExpBuffer(&buf,
- "SELECT c.oid::pg_catalog.regclass, c.relkind, NULL\n"
+ "SELECT c.oid::pg_catalog.regclass, c.relkind, false AS inhdetached, NULL\n"
"FROM pg_catalog.pg_class c, pg_catalog.pg_inherits i\n"
"WHERE c.oid = i.inhrelid AND i.inhparent = '%s'\n"
"ORDER BY c.oid::pg_catalog.regclass::pg_catalog.text;",
oid);
else
printfPQExpBuffer(&buf,
- "SELECT c.oid::pg_catalog.regclass, c.relkind, NULL\n"
+ "SELECT c.oid::pg_catalog.regclass, c.relkind, false AS inhdetached, NULL\n"
"FROM pg_catalog.pg_class c, pg_catalog.pg_inherits i\n"
"WHERE c.oid = i.inhrelid AND i.inhparent = '%s'\n"
"ORDER BY c.relname;",
@@ -3247,11 +3263,13 @@ describeOneTableDetails(const char *schemaname,
else
printfPQExpBuffer(&buf, "%*s %s",
ctw, "", PQgetvalue(result, i, 0));
- if (!PQgetisnull(result, i, 2))
- appendPQExpBuffer(&buf, " %s", PQgetvalue(result, i, 2));
+ if (!PQgetisnull(result, i, 3))
+ appendPQExpBuffer(&buf, " %s", PQgetvalue(result, i, 3));
if (child_relkind == RELKIND_PARTITIONED_TABLE ||
child_relkind == RELKIND_PARTITIONED_INDEX)
appendPQExpBufferStr(&buf, ", PARTITIONED");
+ if (strcmp(PQgetvalue(result, i, 2), "t") == 0)
+ appendPQExpBuffer(&buf, " (DETACHED)");
if (i < tuples - 1)
appendPQExpBufferChar(&buf, ',');
diff --git a/src/include/catalog/pg_inherits.h b/src/include/catalog/pg_inherits.h
index 0432c3ab88..fcdea9f26f 100644
--- a/src/include/catalog/pg_inherits.h
+++ b/src/include/catalog/pg_inherits.h
@@ -34,6 +34,7 @@ CATALOG(pg_inherits,2611,InheritsRelationId)
Oid inhrelid;
Oid inhparent;
int32 inhseqno;
+ bool inhdetached;
} FormData_pg_inherits;
/* ----------------
@@ -44,7 +45,8 @@ CATALOG(pg_inherits,2611,InheritsRelationId)
typedef FormData_pg_inherits *Form_pg_inherits;
-extern List *find_inheritance_children(Oid parentrelId, LOCKMODE lockmode);
+extern List *find_inheritance_children(Oid parentrelId, bool include_detached,
+ LOCKMODE lockmode);
extern List *find_all_inheritors(Oid parentrelId, LOCKMODE lockmode,
List **parents);
extern bool has_subclass(Oid relationId);
@@ -52,6 +54,7 @@ extern bool has_superclass(Oid relationId);
extern bool typeInheritsFrom(Oid subclassTypeId, Oid superclassTypeId);
extern void StoreSingleInheritance(Oid relationId, Oid parentOid,
int32 seqNumber);
-extern bool DeleteInheritsTuple(Oid inhrelid, Oid inhparent);
+extern bool DeleteInheritsTuple(Oid inhrelid, Oid inhparent, bool allow_detached,
+ const char *childname);
#endif /* PG_INHERITS_H */
diff --git a/src/include/nodes/parsenodes.h b/src/include/nodes/parsenodes.h
index 151bcdb7ef..d291b4d7df 100644
--- a/src/include/nodes/parsenodes.h
+++ b/src/include/nodes/parsenodes.h
@@ -859,6 +859,7 @@ typedef struct PartitionCmd
NodeTag type;
RangeVar *name; /* name of partition to attach/detach */
PartitionBoundSpec *bound; /* FOR VALUES, if attaching */
+ bool concurrent;
} PartitionCmd;
/****************************************************************************
@@ -1845,6 +1846,7 @@ typedef enum AlterTableType
AT_GenericOptions, /* OPTIONS (...) */
AT_AttachPartition, /* ATTACH PARTITION */
AT_DetachPartition, /* DETACH PARTITION */
+ AT_DetachPartitionFinalize, /* DETACH PARTITION FINALIZE */
AT_AddIdentity, /* ADD IDENTITY */
AT_SetIdentity, /* SET identity column options */
AT_DropIdentity /* DROP IDENTITY */
diff --git a/src/include/parser/kwlist.h b/src/include/parser/kwlist.h
index 08f22ce211..27414b0912 100644
--- a/src/include/parser/kwlist.h
+++ b/src/include/parser/kwlist.h
@@ -163,6 +163,7 @@ PG_KEYWORD("false", FALSE_P, RESERVED_KEYWORD)
PG_KEYWORD("family", FAMILY, UNRESERVED_KEYWORD)
PG_KEYWORD("fetch", FETCH, RESERVED_KEYWORD)
PG_KEYWORD("filter", FILTER, UNRESERVED_KEYWORD)
+PG_KEYWORD("finalize", FINALIZE, UNRESERVED_KEYWORD)
PG_KEYWORD("first", FIRST_P, UNRESERVED_KEYWORD)
PG_KEYWORD("float", FLOAT_P, COL_NAME_KEYWORD)
PG_KEYWORD("following", FOLLOWING, UNRESERVED_KEYWORD)
diff --git a/src/include/partitioning/partdesc.h b/src/include/partitioning/partdesc.h
index 70df764981..b1ee9fd5e3 100644
--- a/src/include/partitioning/partdesc.h
+++ b/src/include/partitioning/partdesc.h
@@ -21,6 +21,7 @@
typedef struct PartitionDescData
{
int nparts; /* Number of partitions */
+ bool includes_detached; /* Does it include detached partitions */
Oid *oids; /* Array of 'nparts' elements containing
* partition OIDs in order of the their bounds */
bool *is_leaf; /* Array of 'nparts' elements storing whether
@@ -30,9 +31,9 @@ typedef struct PartitionDescData
} PartitionDescData;
-extern PartitionDesc RelationGetPartitionDesc(Relation rel);
+extern PartitionDesc RelationGetPartitionDesc(Relation rel, bool include_detached);
-extern PartitionDirectory CreatePartitionDirectory(MemoryContext mcxt);
+extern PartitionDirectory CreatePartitionDirectory(MemoryContext mcxt, bool include_detached);
extern PartitionDesc PartitionDirectoryLookup(PartitionDirectory, Relation);
extern void DestroyPartitionDirectory(PartitionDirectory pdir);
diff --git a/src/test/isolation/expected/detach-partition-concurrently-1.out b/src/test/isolation/expected/detach-partition-concurrently-1.out
new file mode 100644
index 0000000000..04a0446ccb
--- /dev/null
+++ b/src/test/isolation/expected/detach-partition-concurrently-1.out
@@ -0,0 +1,182 @@
+Parsed test spec with 3 sessions
+
+starting permutation: s1b s1s s2d s1s s1c s2drop s1s
+step s1b: BEGIN;
+step s1s: SELECT * FROM d_listp;
+a
+
+1
+2
+step s2d: ALTER TABLE d_listp DETACH PARTITION d_listp2 CONCURRENTLY; <waiting ...>
+step s1s: SELECT * FROM d_listp;
+a
+
+1
+step s1c: COMMIT;
+step s2d: <... completed>
+step s2drop: DROP TABLE d_listp2;
+step s1s: SELECT * FROM d_listp;
+a
+
+1
+
+starting permutation: s1b s1s s2d s1s s3s s3i s1c s3i s2drop s1s
+step s1b: BEGIN;
+step s1s: SELECT * FROM d_listp;
+a
+
+1
+2
+step s2d: ALTER TABLE d_listp DETACH PARTITION d_listp2 CONCURRENTLY; <waiting ...>
+step s1s: SELECT * FROM d_listp;
+a
+
+1
+step s3s: SELECT * FROM d_listp;
+a
+
+1
+step s3i: SELECT relpartbound IS NULL FROM pg_class where relname = 'd_listp2';
+?column?
+
+f
+step s1c: COMMIT;
+step s2d: <... completed>
+step s3i: SELECT relpartbound IS NULL FROM pg_class where relname = 'd_listp2';
+?column?
+
+t
+step s2drop: DROP TABLE d_listp2;
+step s1s: SELECT * FROM d_listp;
+a
+
+1
+
+starting permutation: s1b s1s s2d s1ins s1c
+step s1b: BEGIN;
+step s1s: SELECT * FROM d_listp;
+a
+
+1
+2
+step s2d: ALTER TABLE d_listp DETACH PARTITION d_listp2 CONCURRENTLY; <waiting ...>
+step s1ins: INSERT INTO d_listp VALUES (1);
+step s1c: COMMIT;
+step s2d: <... completed>
+
+starting permutation: s1brr s1s s2d s1ins s1c
+step s1brr: BEGIN ISOLATION LEVEL REPEATABLE READ;
+step s1s: SELECT * FROM d_listp;
+a
+
+1
+2
+step s2d: ALTER TABLE d_listp DETACH PARTITION d_listp2 CONCURRENTLY; <waiting ...>
+step s1ins: INSERT INTO d_listp VALUES (1);
+step s1c: COMMIT;
+step s2d: <... completed>
+
+starting permutation: s1brr s1s s2d s1s s1c
+step s1brr: BEGIN ISOLATION LEVEL REPEATABLE READ;
+step s1s: SELECT * FROM d_listp;
+a
+
+1
+2
+step s2d: ALTER TABLE d_listp DETACH PARTITION d_listp2 CONCURRENTLY; <waiting ...>
+step s1s: SELECT * FROM d_listp;
+a
+
+1
+step s1c: COMMIT;
+step s2d: <... completed>
+
+starting permutation: s1brr s1prep s1s s2d s1s s1exec s1c
+step s1brr: BEGIN ISOLATION LEVEL REPEATABLE READ;
+step s1prep: PREPARE f(int) AS INSERT INTO d_listp VALUES ($1);
+step s1s: SELECT * FROM d_listp;
+a
+
+1
+2
+step s2d: ALTER TABLE d_listp DETACH PARTITION d_listp2 CONCURRENTLY; <waiting ...>
+step s1s: SELECT * FROM d_listp;
+a
+
+1
+step s1exec: EXECUTE f(1); DEALLOCATE f;
+step s1c: COMMIT;
+step s2d: <... completed>
+
+starting permutation: s1brr s1prep s1s s2d s1s s1exec2 s1c
+step s1brr: BEGIN ISOLATION LEVEL REPEATABLE READ;
+step s1prep: PREPARE f(int) AS INSERT INTO d_listp VALUES ($1);
+step s1s: SELECT * FROM d_listp;
+a
+
+1
+2
+step s2d: ALTER TABLE d_listp DETACH PARTITION d_listp2 CONCURRENTLY; <waiting ...>
+step s1s: SELECT * FROM d_listp;
+a
+
+1
+step s1exec2: EXECUTE f(2); DEALLOCATE f;
+step s2d: <... completed>
+error in steps s1exec2 s2d: ERROR: no partition of relation "d_listp" found for row
+step s1c: COMMIT;
+
+starting permutation: s1b s1s s2d s3drop1 s1c
+step s1b: BEGIN;
+step s1s: SELECT * FROM d_listp;
+a
+
+1
+2
+step s2d: ALTER TABLE d_listp DETACH PARTITION d_listp2 CONCURRENTLY; <waiting ...>
+step s3drop1: DROP TABLE d_listp; <waiting ...>
+step s1c: COMMIT;
+step s2d: <... completed>
+step s3drop1: <... completed>
+error in steps s1c s2d s3drop1: ERROR: partitioned table "d_listp" was removed concurrently
+
+starting permutation: s1b s1s s2d s3drop2 s1c
+step s1b: BEGIN;
+step s1s: SELECT * FROM d_listp;
+a
+
+1
+2
+step s2d: ALTER TABLE d_listp DETACH PARTITION d_listp2 CONCURRENTLY; <waiting ...>
+step s3drop2: DROP TABLE d_listp2; <waiting ...>
+step s1c: COMMIT;
+step s2d: <... completed>
+step s3drop2: <... completed>
+error in steps s1c s2d s3drop2: ERROR: partition "d_listp2" was removed concurrently
+
+starting permutation: s1b s1s s2d s3detach s1c
+step s1b: BEGIN;
+step s1s: SELECT * FROM d_listp;
+a
+
+1
+2
+step s2d: ALTER TABLE d_listp DETACH PARTITION d_listp2 CONCURRENTLY; <waiting ...>
+step s3detach: ALTER TABLE d_listp DETACH PARTITION d_listp2; <waiting ...>
+step s1c: COMMIT;
+step s2d: <... completed>
+step s3detach: <... completed>
+error in steps s1c s2d s3detach: ERROR: cannot detach partition "d_listp2"
+
+starting permutation: s1b s1s s2d s3rename s1c
+step s1b: BEGIN;
+step s1s: SELECT * FROM d_listp;
+a
+
+1
+2
+step s2d: ALTER TABLE d_listp DETACH PARTITION d_listp2 CONCURRENTLY; <waiting ...>
+step s3rename: ALTER TABLE d_listp2 RENAME TO foobar; <waiting ...>
+step s1c: COMMIT;
+step s2d: <... completed>
+step s3rename: <... completed>
diff --git a/src/test/isolation/expected/detach-partition-concurrently-2.out b/src/test/isolation/expected/detach-partition-concurrently-2.out
new file mode 100644
index 0000000000..85be707b40
--- /dev/null
+++ b/src/test/isolation/expected/detach-partition-concurrently-2.out
@@ -0,0 +1,66 @@
+Parsed test spec with 3 sessions
+
+starting permutation: s1b s1s s2d s3i1 s1c
+step s1b: BEGIN;
+step s1s: SELECT * FROM d_lp_fk;
+a
+
+1
+2
+step s2d: ALTER TABLE d_lp_fk DETACH PARTITION d_lp_fk_1 CONCURRENTLY; <waiting ...>
+step s3i1: INSERT INTO d_lp_fk_r VALUES (1);
+ERROR: insert or update on table "d_lp_fk_r" violates foreign key constraint "d_lp_fk_r_a_fkey"
+step s1c: COMMIT;
+step s2d: <... completed>
+
+starting permutation: s1b s1s s2d s3i2 s3i2 s1c
+step s1b: BEGIN;
+step s1s: SELECT * FROM d_lp_fk;
+a
+
+1
+2
+step s2d: ALTER TABLE d_lp_fk DETACH PARTITION d_lp_fk_1 CONCURRENTLY; <waiting ...>
+step s3i2: INSERT INTO d_lp_fk_r VALUES (2);
+step s3i2: INSERT INTO d_lp_fk_r VALUES (2);
+step s1c: COMMIT;
+step s2d: <... completed>
+
+starting permutation: s1b s1s s3i1 s2d s1c
+step s1b: BEGIN;
+step s1s: SELECT * FROM d_lp_fk;
+a
+
+1
+2
+step s3i1: INSERT INTO d_lp_fk_r VALUES (1);
+step s2d: ALTER TABLE d_lp_fk DETACH PARTITION d_lp_fk_1 CONCURRENTLY;
+ERROR: removing partition "d_lp_fk_1" violates foreign key constraint "d_lp_fk_r_a_fkey1"
+step s1c: COMMIT;
+
+starting permutation: s1b s1s s3i2 s2d s1c
+step s1b: BEGIN;
+step s1s: SELECT * FROM d_lp_fk;
+a
+
+1
+2
+step s3i2: INSERT INTO d_lp_fk_r VALUES (2);
+step s2d: ALTER TABLE d_lp_fk DETACH PARTITION d_lp_fk_1 CONCURRENTLY; <waiting ...>
+step s1c: COMMIT;
+step s2d: <... completed>
+
+starting permutation: s1b s1s s3b s2d s3i1 s1c s3c
+step s1b: BEGIN;
+step s1s: SELECT * FROM d_lp_fk;
+a
+
+1
+2
+step s3b: BEGIN;
+step s2d: ALTER TABLE d_lp_fk DETACH PARTITION d_lp_fk_1 CONCURRENTLY; <waiting ...>
+step s3i1: INSERT INTO d_lp_fk_r VALUES (1);
+ERROR: insert or update on table "d_lp_fk_r" violates foreign key constraint "d_lp_fk_r_a_fkey"
+step s1c: COMMIT;
+step s2d: <... completed>
+step s3c: COMMIT;
diff --git a/src/test/isolation/isolation_schedule b/src/test/isolation/isolation_schedule
index 218c87b24b..a9c1abc244 100644
--- a/src/test/isolation/isolation_schedule
+++ b/src/test/isolation/isolation_schedule
@@ -24,6 +24,8 @@ test: deadlock-hard
test: deadlock-soft
test: deadlock-soft-2
test: deadlock-parallel
+test: detach-partition-concurrently-1
+test: detach-partition-concurrently-2
test: fk-contention
test: fk-deadlock
test: fk-deadlock2
diff --git a/src/test/isolation/specs/detach-partition-concurrently-1.spec b/src/test/isolation/specs/detach-partition-concurrently-1.spec
new file mode 100644
index 0000000000..ca92530824
--- /dev/null
+++ b/src/test/isolation/specs/detach-partition-concurrently-1.spec
@@ -0,0 +1,71 @@
+# Test that detach partition concurrently makes the partition invisible at the
+# correct time.
+
+setup
+{
+ DROP TABLE IF EXISTS d_listp, d_listp1, d_listp2;
+ CREATE TABLE d_listp (a int) PARTITION BY LIST(a);
+ CREATE TABLE d_listp1 PARTITION OF d_listp FOR VALUES IN (1);
+ CREATE TABLE d_listp2 PARTITION OF d_listp FOR VALUES IN (2);
+ INSERT INTO d_listp VALUES (1),(2);
+}
+
+teardown {
+ DROP TABLE IF EXISTS d_listp, d_listp2;
+}
+
+session "s1"
+step "s1b" { BEGIN; }
+step "s1brr" { BEGIN ISOLATION LEVEL REPEATABLE READ; }
+step "s1s" { SELECT * FROM d_listp; }
+step "s1ins" { INSERT INTO d_listp VALUES (1); }
+step "s1prep" { PREPARE f(int) AS INSERT INTO d_listp VALUES ($1); }
+step "s1exec" { EXECUTE f(1); DEALLOCATE f; }
+step "s1exec2" { EXECUTE f(2); DEALLOCATE f; }
+step "s1c" { COMMIT; }
+
+session "s2"
+step "s2d" { ALTER TABLE d_listp DETACH PARTITION d_listp2 CONCURRENTLY; }
+step "s2drop" { DROP TABLE d_listp2; }
+
+session "s3"
+step "s3s" { SELECT * FROM d_listp; }
+step "s3i" { SELECT relpartbound IS NULL FROM pg_class where relname = 'd_listp2'; }
+step "s3drop1" { DROP TABLE d_listp; }
+step "s3drop2" { DROP TABLE d_listp2; }
+step "s3detach" { ALTER TABLE d_listp DETACH PARTITION d_listp2; }
+step "s3rename" { ALTER TABLE d_listp2 RENAME TO foobar; }
+
+# The transaction that detaches hangs until it sees any older transaction
+# terminate.
+permutation "s1b" "s1s" "s2d" "s1s" "s1c" "s2drop" "s1s"
+
+# relpartbound remains set until s1 commits
+# XXX this could be timing dependent :-(
+permutation "s1b" "s1s" "s2d" "s1s" "s3s" "s3i" "s1c" "s3i" "s2drop" "s1s"
+
+# Session s1 no longer sees the partition ...
+permutation "s1b" "s1s" "s2d" "s1ins" "s1c"
+# ... regardless of the isolation level.
+permutation "s1brr" "s1s" "s2d" "s1ins" "s1c"
+# However, DETACH CONCURRENTLY is not isolation-level-correct
+permutation "s1brr" "s1s" "s2d" "s1s" "s1c"
+
+# a prepared query is not blocked
+permutation "s1brr" "s1prep" "s1s" "s2d" "s1s" "s1exec" "s1c"
+permutation "s1brr" "s1prep" "s1s" "s2d" "s1s" "s1exec2" "s1c"
+
+
+# Note: the following tests are not very interesting in isolationtester because
+# the way this tool handles multiple sessions: it'll only run the drop/detach/
+# rename commands after s2 is blocked waiting. It would be more useful to
+# run the DDL when s2 commits its first transaction instead. Maybe these should
+# become TAP tests someday.
+
+# What happens if the partition or the parent table is dropped while waiting?
+permutation "s1b" "s1s" "s2d" "s3drop1" "s1c"
+permutation "s1b" "s1s" "s2d" "s3drop2" "s1c"
+# Also try a non-concurrent detach while the other one is waiting
+permutation "s1b" "s1s" "s2d" "s3detach" "s1c"
+# and some other DDL
+permutation "s1b" "s1s" "s2d" "s3rename" "s1c"
diff --git a/src/test/isolation/specs/detach-partition-concurrently-2.spec b/src/test/isolation/specs/detach-partition-concurrently-2.spec
new file mode 100644
index 0000000000..9281c80a69
--- /dev/null
+++ b/src/test/isolation/specs/detach-partition-concurrently-2.spec
@@ -0,0 +1,41 @@
+# Test that detach partition concurrently makes the partition safe
+# for foreign keys that reference it.
+
+setup
+{
+ DROP TABLE IF EXISTS d_lp_fk, d_lp_fk_1, d_lp_fk_2, d_lp_fk_r;
+
+ CREATE TABLE d_lp_fk (a int PRIMARY KEY) PARTITION BY LIST(a);
+ CREATE TABLE d_lp_fk_1 PARTITION OF d_lp_fk FOR VALUES IN (1);
+ CREATE TABLE d_lp_fk_2 PARTITION OF d_lp_fk FOR VALUES IN (2);
+ INSERT INTO d_lp_fk VALUES (1), (2);
+
+ CREATE TABLE d_lp_fk_r (a int references d_lp_fk);
+}
+
+teardown { DROP TABLE IF EXISTS d_lp_fk, d_lp_fk_1, d_lp_fk_2, d_lp_fk_r; }
+
+session "s1"
+step "s1b" { BEGIN; }
+step "s1s" { SELECT * FROM d_lp_fk; }
+step "s1c" { COMMIT; }
+
+session "s2"
+step "s2d" { ALTER TABLE d_lp_fk DETACH PARTITION d_lp_fk_1 CONCURRENTLY; }
+
+session "s3"
+step "s3b" { BEGIN; }
+step "s3i1" { INSERT INTO d_lp_fk_r VALUES (1); }
+step "s3i2" { INSERT INTO d_lp_fk_r VALUES (2); }
+step "s3c" { COMMIT; }
+
+# The transaction that detaches hangs until it sees any older transaction
+# terminate.
+permutation "s1b" "s1s" "s2d" "s3i1" "s1c"
+permutation "s1b" "s1s" "s2d" "s3i2" "s3i2" "s1c"
+
+permutation "s1b" "s1s" "s3i1" "s2d" "s1c"
+permutation "s1b" "s1s" "s3i2" "s2d" "s1c"
+
+# what if s3 has an uncommitted insertion?
+permutation "s1b" "s1s" "s3b" "s2d" "s3i1" "s1c" "s3c"
--
2.20.1
--ikeVEW9yuYc//A+q--
^ permalink raw reply [nested|flat] 5+ messages in thread
* speed up a logical replica setup
@ 2022-02-21 12:09 Euler Taveira <[email protected]>
0 siblings, 1 reply; 5+ messages in thread
From: Euler Taveira @ 2022-02-21 12:09 UTC (permalink / raw)
To: [email protected]
Hi,
Logical replication has been used to migration with minimal downtime. However,
if you are dealing with a big database, the amount of required resources (disk
-- due to WAL retention) increases as the backlog (WAL) increases. Unless you
have a generous amount of resources and can wait for long period of time until
the new replica catches up, creating a logical replica is impracticable on
large databases.
The general idea is to create and convert a physical replica or a base backup
(archived WAL files available) into a logical replica. The initial data copy
and catchup tends to be faster on a physical replica. This technique has been
successfully used in pglogical_create_subscriber [1].
A new tool called pg_subscriber does this conversion and is tightly integrated
with Postgres.
DESIGN
The conversion requires 8 steps.
1. Check if the target data directory has the same system identifier than the
source data directory.
2. Stop the target server if it is running as a standby server. (Modify
recovery parameters requires a restart.)
3. Create one replication slot per specified database on the source server. One
additional replication slot is created at the end to get the consistent LSN
(This consistent LSN will be used as (a) a stopping point for the recovery
process and (b) a starting point for the subscriptions).
4. Write recovery parameters into the target data directory and start the
target server (Wait until the target server is promoted).
5. Create one publication (FOR ALL TABLES) per specified database on the source
server.
6. Create one subscription per specified database on the target server (Use
replication slot and publication created in a previous step. Don't enable the
subscriptions yet).
7. Sets the replication progress to the consistent LSN that was got in a
previous step.
8. Enable the subscription for each specified database on the target server.
This tool does not take a base backup. It can certainly be included later.
There is already a tool do it: pg_basebackup.
There is a --subscriber-conninfo option to inform the subscriber connection
string, however, we could remove it since this tool runs on the subscriber and
we can build a connection string.
NAME
I'm not sure about the proposed name. I came up with this one because it is not
so long. The last added tools uses pg_ prefix, verb (action) and object.
pg_initsubscriber and pg_createsubscriber are names that I thought but I'm not
excited about it.
DOCUMENTATION
It is available and describes this tool.
TESTS
Basic tests are included. It requires some tests to exercise this tool.
Comments?
[1] https://github.com/2ndQuadrant/pglogical
--
Euler Taveira
EDB https://www.enterprisedb.com/
Attachments:
[text/x-patch] v1-0001-Move-readfile-and-free_readfile-to-file_utils.h.patch (10.9K, ../../[email protected]/3-v1-0001-Move-readfile-and-free_readfile-to-file_utils.h.patch)
download | inline diff:
From 5827245b8a06f906f603100c8fb27be533a6c0a1 Mon Sep 17 00:00:00 2001
From: Euler Taveira <[email protected]>
Date: Fri, 11 Feb 2022 03:05:58 -0300
Subject: [PATCH v1 1/2] Move readfile() and free_readfile() to file_utils.h
Allow these functions to be used by other binaries.
There is a static function called readfile() in initdb.c too. Rename it
to avoid conflicting with the exposed function.
---
src/bin/initdb/initdb.c | 26 +++----
src/bin/pg_ctl/pg_ctl.c | 122 +-------------------------------
src/common/file_utils.c | 119 +++++++++++++++++++++++++++++++
src/include/common/file_utils.h | 3 +
4 files changed, 136 insertions(+), 134 deletions(-)
diff --git a/src/bin/initdb/initdb.c b/src/bin/initdb/initdb.c
index 97f15971e2..a4cbfeb954 100644
--- a/src/bin/initdb/initdb.c
+++ b/src/bin/initdb/initdb.c
@@ -243,8 +243,8 @@ static char **replace_token(char **lines,
#ifndef HAVE_UNIX_SOCKETS
static char **filter_lines_with_token(char **lines, const char *token);
#endif
-static char **readfile(const char *path);
-static void writefile(char *path, char **lines);
+static char **read_text_file(const char *path);
+static void write_text_file(char *path, char **lines);
static FILE *popen_check(const char *command, const char *mode);
static char *get_id(void);
static int get_encoding_id(const char *encoding_name);
@@ -453,7 +453,7 @@ filter_lines_with_token(char **lines, const char *token)
* get the lines from a text file
*/
static char **
-readfile(const char *path)
+read_text_file(const char *path)
{
char **result;
FILE *infile;
@@ -500,7 +500,7 @@ readfile(const char *path)
* so that the resulting configuration files are nicely editable on Windows.
*/
static void
-writefile(char *path, char **lines)
+write_text_file(char *path, char **lines)
{
FILE *out_file;
char **line;
@@ -1063,7 +1063,7 @@ setup_config(void)
/* postgresql.conf */
- conflines = readfile(conf_file);
+ conflines = read_text_file(conf_file);
snprintf(repltok, sizeof(repltok), "max_connections = %d", n_connections);
conflines = replace_token(conflines, "#max_connections = 100", repltok);
@@ -1214,7 +1214,7 @@ setup_config(void)
snprintf(path, sizeof(path), "%s/postgresql.conf", pg_data);
- writefile(path, conflines);
+ write_text_file(path, conflines);
if (chmod(path, pg_file_create_mode) != 0)
{
pg_log_error("could not change permissions of \"%s\": %m", path);
@@ -1233,7 +1233,7 @@ setup_config(void)
sprintf(path, "%s/postgresql.auto.conf", pg_data);
- writefile(path, autoconflines);
+ write_text_file(path, autoconflines);
if (chmod(path, pg_file_create_mode) != 0)
{
pg_log_error("could not change permissions of \"%s\": %m", path);
@@ -1245,7 +1245,7 @@ setup_config(void)
/* pg_hba.conf */
- conflines = readfile(hba_file);
+ conflines = read_text_file(hba_file);
#ifndef HAVE_UNIX_SOCKETS
conflines = filter_lines_with_token(conflines, "@remove-line-for-nolocal@");
@@ -1319,7 +1319,7 @@ setup_config(void)
snprintf(path, sizeof(path), "%s/pg_hba.conf", pg_data);
- writefile(path, conflines);
+ write_text_file(path, conflines);
if (chmod(path, pg_file_create_mode) != 0)
{
pg_log_error("could not change permissions of \"%s\": %m", path);
@@ -1330,11 +1330,11 @@ setup_config(void)
/* pg_ident.conf */
- conflines = readfile(ident_file);
+ conflines = read_text_file(ident_file);
snprintf(path, sizeof(path), "%s/pg_ident.conf", pg_data);
- writefile(path, conflines);
+ write_text_file(path, conflines);
if (chmod(path, pg_file_create_mode) != 0)
{
pg_log_error("could not change permissions of \"%s\": %m", path);
@@ -1362,7 +1362,7 @@ bootstrap_template1(void)
printf(_("running bootstrap script ... "));
fflush(stdout);
- bki_lines = readfile(bki_file);
+ bki_lines = read_text_file(bki_file);
/* Check that bki file appears to be of the right version */
@@ -1547,7 +1547,7 @@ setup_run_file(FILE *cmdfd, const char *filename)
{
char **lines;
- lines = readfile(filename);
+ lines = read_text_file(filename);
for (char **line = lines; *line != NULL; line++)
{
diff --git a/src/bin/pg_ctl/pg_ctl.c b/src/bin/pg_ctl/pg_ctl.c
index 3c182c97d4..b87beb7380 100644
--- a/src/bin/pg_ctl/pg_ctl.c
+++ b/src/bin/pg_ctl/pg_ctl.c
@@ -26,6 +26,7 @@
#include "catalog/pg_control.h"
#include "common/controldata_utils.h"
#include "common/file_perm.h"
+#include "common/file_utils.h"
#include "common/logging.h"
#include "common/string.h"
#include "getopt_long.h"
@@ -150,8 +151,6 @@ static PTOKEN_PRIVILEGES GetPrivilegesToDelete(HANDLE hToken);
#endif
static pgpid_t get_pgpid(bool is_status_request);
-static char **readfile(const char *path, int *numlines);
-static void free_readfile(char **optlines);
static pgpid_t start_postmaster(void);
static void read_post_opts(void);
@@ -307,125 +306,6 @@ get_pgpid(bool is_status_request)
}
-/*
- * get the lines from a text file - return NULL if file can't be opened
- *
- * Trailing newlines are deleted from the lines (this is a change from pre-v10)
- *
- * *numlines is set to the number of line pointers returned; there is
- * also an additional NULL pointer after the last real line.
- */
-static char **
-readfile(const char *path, int *numlines)
-{
- int fd;
- int nlines;
- char **result;
- char *buffer;
- char *linebegin;
- int i;
- int n;
- int len;
- struct stat statbuf;
-
- *numlines = 0; /* in case of failure or empty file */
-
- /*
- * Slurp the file into memory.
- *
- * The file can change concurrently, so we read the whole file into memory
- * with a single read() call. That's not guaranteed to get an atomic
- * snapshot, but in practice, for a small file, it's close enough for the
- * current use.
- */
- fd = open(path, O_RDONLY | PG_BINARY, 0);
- if (fd < 0)
- return NULL;
- if (fstat(fd, &statbuf) < 0)
- {
- close(fd);
- return NULL;
- }
- if (statbuf.st_size == 0)
- {
- /* empty file */
- close(fd);
- result = (char **) pg_malloc(sizeof(char *));
- *result = NULL;
- return result;
- }
- buffer = pg_malloc(statbuf.st_size + 1);
-
- len = read(fd, buffer, statbuf.st_size + 1);
- close(fd);
- if (len != statbuf.st_size)
- {
- /* oops, the file size changed between fstat and read */
- free(buffer);
- return NULL;
- }
-
- /*
- * Count newlines. We expect there to be a newline after each full line,
- * including one at the end of file. If there isn't a newline at the end,
- * any characters after the last newline will be ignored.
- */
- nlines = 0;
- for (i = 0; i < len; i++)
- {
- if (buffer[i] == '\n')
- nlines++;
- }
-
- /* set up the result buffer */
- result = (char **) pg_malloc((nlines + 1) * sizeof(char *));
- *numlines = nlines;
-
- /* now split the buffer into lines */
- linebegin = buffer;
- n = 0;
- for (i = 0; i < len; i++)
- {
- if (buffer[i] == '\n')
- {
- int slen = &buffer[i] - linebegin;
- char *linebuf = pg_malloc(slen + 1);
-
- memcpy(linebuf, linebegin, slen);
- /* we already dropped the \n, but get rid of any \r too */
- if (slen > 0 && linebuf[slen - 1] == '\r')
- slen--;
- linebuf[slen] = '\0';
- result[n++] = linebuf;
- linebegin = &buffer[i + 1];
- }
- }
- result[n] = NULL;
-
- free(buffer);
-
- return result;
-}
-
-
-/*
- * Free memory allocated for optlines through readfile()
- */
-static void
-free_readfile(char **optlines)
-{
- char *curr_line = NULL;
- int i = 0;
-
- if (!optlines)
- return;
-
- while ((curr_line = optlines[i++]))
- free(curr_line);
-
- free(optlines);
-}
-
/*
* start/test/stop routines
*/
diff --git a/src/common/file_utils.c b/src/common/file_utils.c
index 7138068633..1f53f088c6 100644
--- a/src/common/file_utils.c
+++ b/src/common/file_utils.c
@@ -398,6 +398,125 @@ durable_rename(const char *oldfile, const char *newfile)
return 0;
}
+/*
+ * get the lines from a text file - return NULL if file can't be opened
+ *
+ * Trailing newlines are deleted from the lines (this is a change from pre-v10)
+ *
+ * *numlines is set to the number of line pointers returned; there is
+ * also an additional NULL pointer after the last real line.
+ */
+char **
+readfile(const char *path, int *numlines)
+{
+ int fd;
+ int nlines;
+ char **result;
+ char *buffer;
+ char *linebegin;
+ int i;
+ int n;
+ int len;
+ struct stat statbuf;
+
+ *numlines = 0; /* in case of failure or empty file */
+
+ /*
+ * Slurp the file into memory.
+ *
+ * The file can change concurrently, so we read the whole file into memory
+ * with a single read() call. That's not guaranteed to get an atomic
+ * snapshot, but in practice, for a small file, it's close enough for the
+ * current use.
+ */
+ fd = open(path, O_RDONLY | PG_BINARY, 0);
+ if (fd < 0)
+ return NULL;
+ if (fstat(fd, &statbuf) < 0)
+ {
+ close(fd);
+ return NULL;
+ }
+ if (statbuf.st_size == 0)
+ {
+ /* empty file */
+ close(fd);
+ result = (char **) pg_malloc(sizeof(char *));
+ *result = NULL;
+ return result;
+ }
+ buffer = pg_malloc(statbuf.st_size + 1);
+
+ len = read(fd, buffer, statbuf.st_size + 1);
+ close(fd);
+ if (len != statbuf.st_size)
+ {
+ /* oops, the file size changed between fstat and read */
+ free(buffer);
+ return NULL;
+ }
+
+ /*
+ * Count newlines. We expect there to be a newline after each full line,
+ * including one at the end of file. If there isn't a newline at the end,
+ * any characters after the last newline will be ignored.
+ */
+ nlines = 0;
+ for (i = 0; i < len; i++)
+ {
+ if (buffer[i] == '\n')
+ nlines++;
+ }
+
+ /* set up the result buffer */
+ result = (char **) pg_malloc((nlines + 1) * sizeof(char *));
+ *numlines = nlines;
+
+ /* now split the buffer into lines */
+ linebegin = buffer;
+ n = 0;
+ for (i = 0; i < len; i++)
+ {
+ if (buffer[i] == '\n')
+ {
+ int slen = &buffer[i] - linebegin;
+ char *linebuf = pg_malloc(slen + 1);
+
+ memcpy(linebuf, linebegin, slen);
+ /* we already dropped the \n, but get rid of any \r too */
+ if (slen > 0 && linebuf[slen - 1] == '\r')
+ slen--;
+ linebuf[slen] = '\0';
+ result[n++] = linebuf;
+ linebegin = &buffer[i + 1];
+ }
+ }
+ result[n] = NULL;
+
+ free(buffer);
+
+ return result;
+}
+
+
+/*
+ * Free memory allocated for optlines through readfile()
+ */
+void
+free_readfile(char **optlines)
+{
+ char *curr_line = NULL;
+ int i = 0;
+
+ if (!optlines)
+ return;
+
+ while ((curr_line = optlines[i++]))
+ free(curr_line);
+
+ free(optlines);
+}
+
#endif /* FRONTEND */
/*
diff --git a/src/include/common/file_utils.h b/src/include/common/file_utils.h
index 2811744c12..27736e3dd7 100644
--- a/src/include/common/file_utils.h
+++ b/src/include/common/file_utils.h
@@ -30,6 +30,9 @@ extern void fsync_pgdata(const char *pg_data, int serverVersion);
extern void fsync_dir_recurse(const char *dir);
extern int durable_rename(const char *oldfile, const char *newfile);
extern int fsync_parent_path(const char *fname);
+
+extern char **readfile(const char *path, int *numlines);
+extern void free_readfile(char **optlines);
#endif
extern PGFileType get_dirent_type(const char *path,
--
2.30.2
[text/x-patch] v1-0002-Create-a-new-logical-replica-from-a-base-backup-o.patch (55.2K, ../../[email protected]/4-v1-0002-Create-a-new-logical-replica-from-a-base-backup-o.patch)
download | inline diff:
From 81788de158abbe1ffb378ffb0884b943232e51b0 Mon Sep 17 00:00:00 2001
From: Euler Taveira <[email protected]>
Date: Fri, 11 Feb 2022 03:17:57 -0300
Subject: [PATCH v1 2/2] Create a new logical replica from a base backup or
standby server.
A new tool called pg_subscriber can convert a physical replica or a base
backup into a logical replica. It runs on the target server and should
be able to connect to the source server (publisher) and the target
server (subscriber).
The conversion requires eight steps. Check if the target data directory
ha the same system identifier than the source data directory. Stop the
target server if it is running as a standby server. Create one
replication slot per specified database on the source server. One
additional replication slot is created at the end to get the consistent
LSN (This consistent LSN will be used as (a) a stopping point for the
recovery process and (b) a starting point for the subscriptions). Write
recovery parameters into the target data directory and start the target
server (Wait until the target server is promoted). Create one
publication (FOR ALL TABLES) per specified database on the source
server. Create one subscription per specified database on the target
server (Use replication slot and publication created in a previous step.
Don't enable the subscriptions yet). Sets the replication progress to
the consistent LSN that was got in a previous step. Enable the
subscription for each specified database on the target server.
Depending on your workload and database size, creating a logical replica
couldn't be an option due to resource contraints (WAL backlog should be
available until all table data is synchronized). The initial data copy
and the replication progress tends to be faster on a physical replica.
The purpose of this tool is to speed up a logical replica setup.
---
doc/src/sgml/ref/allfiles.sgml | 1 +
doc/src/sgml/ref/pg_subscriber.sgml | 242 ++++
doc/src/sgml/reference.sgml | 1 +
src/bin/Makefile | 1 +
src/bin/pg_subscriber/Makefile | 39 +
src/bin/pg_subscriber/pg_subscriber.c | 1463 +++++++++++++++++++++++++
src/bin/pg_subscriber/t/001_basic.pl | 41 +
src/tools/msvc/Mkvcbuild.pm | 2 +-
8 files changed, 1789 insertions(+), 1 deletion(-)
create mode 100644 doc/src/sgml/ref/pg_subscriber.sgml
create mode 100644 src/bin/pg_subscriber/Makefile
create mode 100644 src/bin/pg_subscriber/pg_subscriber.c
create mode 100644 src/bin/pg_subscriber/t/001_basic.pl
diff --git a/doc/src/sgml/ref/allfiles.sgml b/doc/src/sgml/ref/allfiles.sgml
index d67270ccc3..eab7f2f616 100644
--- a/doc/src/sgml/ref/allfiles.sgml
+++ b/doc/src/sgml/ref/allfiles.sgml
@@ -212,6 +212,7 @@ Complete list of usable sgml source files in this directory.
<!ENTITY pgResetwal SYSTEM "pg_resetwal.sgml">
<!ENTITY pgRestore SYSTEM "pg_restore.sgml">
<!ENTITY pgRewind SYSTEM "pg_rewind.sgml">
+<!ENTITY pgSubscriber SYSTEM "pg_subscriber.sgml">
<!ENTITY pgVerifyBackup SYSTEM "pg_verifybackup.sgml">
<!ENTITY pgtestfsync SYSTEM "pgtestfsync.sgml">
<!ENTITY pgtesttiming SYSTEM "pgtesttiming.sgml">
diff --git a/doc/src/sgml/ref/pg_subscriber.sgml b/doc/src/sgml/ref/pg_subscriber.sgml
new file mode 100644
index 0000000000..e68a19092e
--- /dev/null
+++ b/doc/src/sgml/ref/pg_subscriber.sgml
@@ -0,0 +1,242 @@
+<!--
+doc/src/sgml/ref/pg_subscriber.sgml
+PostgreSQL documentation
+-->
+
+<refentry id="app-pgsubscriber">
+ <indexterm zone="app-pgsubscriber">
+ <primary>pg_subscriber</primary>
+ </indexterm>
+
+ <refmeta>
+ <refentrytitle><application>pg_subscriber</application></refentrytitle>
+ <manvolnum>1</manvolnum>
+ <refmiscinfo>Application</refmiscinfo>
+ </refmeta>
+
+ <refnamediv>
+ <refname>pg_subscriber</refname>
+ <refpurpose>create a new logical replica from a base backup of a
+ <productname>PostgreSQL</productname> cluster or a standby
+ server</refpurpose>
+ </refnamediv>
+
+ <refsynopsisdiv>
+ <cmdsynopsis>
+ <command>pg_subscriber</command>
+ <arg rep="repeat"><replaceable>option</replaceable></arg>
+ </cmdsynopsis>
+ </refsynopsisdiv>
+
+ <refsect1>
+ <title>Description</title>
+ <para>
+ <application>pg_subscriber</application> takes the publisher and subscriber
+ connection strings, a base backup directory and a list of database names and
+ it sets up a new logical replica using the physical recovery process. A
+ standby server can also be used.
+ </para>
+
+ <para>
+ The <application>pg_subscriber</application> should be run at the target
+ server. The source server (known as publisher server) should accept logical
+ replication connections from the target server (known as subscriber server).
+ The target server should accept local logical replication connection.
+ </para>
+
+ <para>
+ The transformation proceeds in eight steps. First,
+ <application>pg_subscriber</application> checks if the given target data
+ directory has the same system identifier than the source data directory.
+ Since it uses the recovery process as one of the steps, it starts the target
+ server as a replica from the source server. If the system identifier is not
+ the same, <application>pg_subscriber</application> will terminate with an
+ error.
+ </para>
+
+ <para>
+ Second, <application>pg_subscriber</application> checks if the target data
+ directory is used by a standby server. Stop the standby server if it is
+ running. One of the next steps is to add some recovery parameters that
+ requires a server start. This step avoids an error.
+ </para>
+
+ <para>
+ Next, <application>pg_subscriber</application> creates one replication slot
+ for each specified database on the source server. The replication slot name
+ contains a <literal>pg_subscriber</literal> prefix. These replication slots
+ will be used by the subscriptions in a future step. Another replication
+ slot is used to get a consistent start location. This consistent LSN will be
+ used (a) as a stopping point in the <xref
+ linkend="guc-recovery-target-lsn"/> parameter and (b) by the subscriptions
+ as a replication starting point. It guarantees that no transaction will be
+ lost.
+ </para>
+
+ <para>
+ Next, write recovery parameters into the target data directory and start the
+ target server. It specifies a LSN (consistent LSN that was obtained in the
+ previous step) of write-ahead log location up to which recovery will
+ proceed. It also specifies <literal>promote</literal> as the action that the
+ server should take once the recovery target is reached. This step finishes
+ once the server ends standby mode and is accepting read-write operations.
+ </para>
+
+ <para>
+ Next, <application>pg_subscriber</application> creates one publication for
+ each specified database on the source server. Each publication replicates
+ changes for all tables in the database. The publication name contains a
+ <literal>pg_subscriber</literal> prefix. These publication will be used by a
+ corresponding subscription in a next step.
+ </para>
+
+ <para>
+ Next, <application>pg_subscriber</application> creates one subscription for
+ each specified database on the target server. Each subscription name
+ contains a <literal>pg_subscriber</literal> prefix. The replication slot
+ name is identical to the subscription name. It also does not copy existing
+ data from the source server. It does not create a replication slot. Instead,
+ it uses the replication slot that was created in a previous step. The
+ subscription is created but it is not enabled yet. The reason is the
+ replication progress must be set to the consistent LSN but replication
+ origin name contains the subscription oid in its name. Hence, the
+ subscription will be enabled in a separate step.
+ </para>
+
+ <para>
+ Next, <application>pg_subscriber</application> sets the replication progress
+ to the consistent LSN that was obtained in a previous step. When the target
+ server started the recovery process, it caught up to the consistent LSN.
+ This is the exact LSN to be used as a initial location for the logical
+ replication.
+ </para>
+
+ <para>
+ Finally, <application>pg_subscriber</application> enables the subscription
+ for each specified database on the target server. The subscription starts
+ streaming from the consistent LSN.
+ </para>
+ </refsect1>
+
+ <refsect1>
+ <title>Options</title>
+
+ <para>
+ <application>pg_subscriber</application> accepts the following
+ command-line arguments:
+
+ <variablelist>
+ <varlistentry>
+ <term><option>-D <replaceable class="parameter">directory</replaceable></option></term>
+ <term><option>--pgdata=<replaceable class="parameter">directory</replaceable></option></term>
+ <listitem>
+ <para>
+ The target directory that contains a base backup. It can also be a
+ cluster directory from a standby server.
+ </para>
+ </listitem>
+ </varlistentry>
+ <varlistentry>
+ <term><option>-P <replaceable class="parameter">conninfo</replaceable></option></term>
+ <term><option>--publisher-conninfo=<replaceable class="parameter">conninfo</replaceable></option></term>
+ <listitem>
+ <para>
+ The connection string to the publisher. For details see <xref linkend="libpq-connstring"/>.
+ </para>
+ </listitem>
+ </varlistentry>
+ <varlistentry>
+ <term><option>-S <replaceable class="parameter">conninfo</replaceable></option></term>
+ <term><option>--subscriber-conninfo=<replaceable class="parameter">conninfo</replaceable></option></term>
+ <listitem>
+ <para>
+ The connection string to the subscriber. For details see <xref linkend="libpq-connstring"/>.
+ </para>
+ </listitem>
+ </varlistentry>
+ <varlistentry>
+ <term><option>-d <replaceable class="parameter">dbname</replaceable></option></term>
+ <term><option>--database=<replaceable class="parameter">dbname</replaceable></option></term>
+ <listitem>
+ <para>
+ The database name to create the subscription. Multiple databases can be
+ selected by writing multiple <option>-d</option> switches.
+ </para>
+ </listitem>
+ </varlistentry>
+
+ <varlistentry>
+ <term><option>-v</option></term>
+ <term><option>--verbose</option></term>
+ <listitem>
+ <para>
+ Enables verbose mode. This will cause
+ <application>pg_subscriber</application> to output progress messages
+ and detailed information about each step.
+ </para>
+ </listitem>
+ </varlistentry>
+ </variablelist>
+ </para>
+
+ <para>
+ Other options are also available:
+
+ <variablelist>
+ <varlistentry>
+ <term><option>-V</option></term>
+ <term><option>--version</option></term>
+ <listitem>
+ <para>
+ Print the <application>pg_subscriber</application> version and exit.
+ </para>
+ </listitem>
+ </varlistentry>
+
+ <varlistentry>
+ <term><option>-?</option></term>
+ <term><option>--help</option></term>
+ <listitem>
+ <para>
+ Show help about <application>pg_subscriber</application> command
+ line arguments, and exit.
+ </para>
+ </listitem>
+ </varlistentry>
+
+ </variablelist>
+ </para>
+
+ </refsect1>
+
+ <refsect1>
+ <title>Examples</title>
+
+ <para>
+ To create a logical replica for database <literal>bar</literal> from a base
+ backup of the server at <literal>foo</literal>:
+<screen>
+<prompt>$</prompt> <userinput>pg_basebackup -h foo -D /usr/local/pgsql/data</userinput>
+<prompt>$</prompt> <userinput>pg_subscriber -D /usr/local/pgsql/data -P "host=foo" -S "host=localhost" -d bar</userinput>
+</screen>
+ </para>
+
+ <para>
+ To create a logical replica for databases <literal>hr</literal> and
+ <literal>finance</literal> from a standby server at <literal>foo</literal>:
+<screen>
+<prompt>$</prompt> <userinput>pg_subscriber -D /usr/local/pgsql/data -P "host=foo" -S "host=localhost" -d hr -d finance</userinput>
+</screen>
+ </para>
+
+ </refsect1>
+
+ <refsect1>
+ <title>See Also</title>
+
+ <simplelist type="inline">
+ <member><xref linkend="app-pgbasebackup"/></member>
+ </simplelist>
+ </refsect1>
+
+</refentry>
diff --git a/doc/src/sgml/reference.sgml b/doc/src/sgml/reference.sgml
index da421ff24e..3566c6050c 100644
--- a/doc/src/sgml/reference.sgml
+++ b/doc/src/sgml/reference.sgml
@@ -256,6 +256,7 @@
&pgReceivewal;
&pgRecvlogical;
&pgRestore;
+ &pgSubscriber;
&pgVerifyBackup;
&psqlRef;
&reindexdb;
diff --git a/src/bin/Makefile b/src/bin/Makefile
index 7f9dde924e..6c4d3c1ffe 100644
--- a/src/bin/Makefile
+++ b/src/bin/Makefile
@@ -25,6 +25,7 @@ SUBDIRS = \
pg_dump \
pg_resetwal \
pg_rewind \
+ pg_subscriber \
pg_test_fsync \
pg_test_timing \
pg_upgrade \
diff --git a/src/bin/pg_subscriber/Makefile b/src/bin/pg_subscriber/Makefile
new file mode 100644
index 0000000000..c48dca6e49
--- /dev/null
+++ b/src/bin/pg_subscriber/Makefile
@@ -0,0 +1,39 @@
+# src/bin/pg_subscriber/Makefile
+
+PGFILEDESC = "pg_subscriber - create a new logical replica from a base backup or a standby server"
+PGAPPICON=win32
+
+subdir = src/bin/pg_subscriber
+top_builddir = ../../..
+include $(top_builddir)/src/Makefile.global
+
+override CPPFLAGS := -I$(libpq_srcdir) $(CPPFLAGS)
+LDFLAGS_INTERNAL += -L$(top_builddir)/src/fe_utils -lpgfeutils $(libpq_pgport)
+
+OBJS = \
+ $(WIN32RES) \
+ pg_subscriber.o
+
+all: pg_subscriber
+
+pg_subscriber: $(OBJS) | submake-libpq submake-libpgport submake-libpgfeutils
+ $(CC) $(CFLAGS) $^ $(LDFLAGS) $(LDFLAGS_EX) $(LIBS) -o $@$(X)
+
+install: all installdirs
+ $(INSTALL_PROGRAM) pg_subscriber$(X) '$(DESTDIR)$(bindir)/pg_subscriber$(X)'
+
+installdirs:
+ $(MKDIR_P) '$(DESTDIR)$(bindir)'
+
+uninstall:
+ rm -f '$(DESTDIR)$(bindir)/pg_subscriber$(X)'
+
+clean distclean maintainer-clean:
+ rm -f pg_subscriber$(X) $(OBJS)
+ rm -rf tmp_check
+
+check:
+ $(prove_check)
+
+installcheck:
+ $(prove_installcheck)
diff --git a/src/bin/pg_subscriber/pg_subscriber.c b/src/bin/pg_subscriber/pg_subscriber.c
new file mode 100644
index 0000000000..7565950a08
--- /dev/null
+++ b/src/bin/pg_subscriber/pg_subscriber.c
@@ -0,0 +1,1463 @@
+/*-------------------------------------------------------------------------
+ *
+ * pg_subscriber.c
+ * Create a new logical replica from a base backup or a standby server
+ *
+ * Copyright (C) 2022, PostgreSQL Global Development Group
+ *
+ * IDENTIFICATION
+ * src/bin/pg_subscriber/pg_subscriber.c
+ *
+ *-------------------------------------------------------------------------
+ */
+#include "postgres_fe.h"
+
+#include <signal.h>
+#include <sys/stat.h>
+#include <sys/wait.h>
+#include <time.h>
+
+#include "catalog/pg_control.h"
+#include "common/connect.h"
+#include "common/controldata_utils.h"
+#include "common/file_utils.h"
+#include "common/logging.h"
+#include "fe_utils/recovery_gen.h"
+#include "fe_utils/simple_list.h"
+#include "getopt_long.h"
+#include "utils/pidfile.h"
+
+typedef struct LogicalRepInfo
+{
+ Oid oid; /* database OID */
+ char *dbname; /* database name */
+ char *pubconninfo; /* publication connection string for logical
+ * replication */
+ char *subconninfo; /* subscription connection string for logical
+ * replication */
+ char *pubname; /* publication name */
+ char *subname; /* subscription name (also replication slot
+ * name) */
+
+ bool made_replslot; /* replication slot was created */
+ bool made_publication; /* publication was created */
+ bool made_subscription; /* subscription was created */
+} LogicalRepInfo;
+
+static void cleanup_objects_atexit(void);
+static void usage();
+static char *get_base_conninfo(char *conninfo, char *dbname,
+ const char *noderole);
+static bool check_data_directory(const char *datadir);
+static char *concat_conninfo_dbname(const char *conninfo, const char *dbname);
+static PGconn *connect_database(const char *conninfo, bool secure_search_path);
+static void disconnect_database(PGconn *conn);
+static char *get_sysid_from_conn(const char *conninfo);
+static char *get_control_from_datadir(const char *datadir);
+static char *create_logical_replication_slot(PGconn *conn, LogicalRepInfo *dbinfo,
+ const char *slot_name);
+static void drop_replication_slot(PGconn *conn, LogicalRepInfo *dbinfo, const char *slot_name);
+static void pg_ctl_status(const char *pg_ctl_cmd, int rc, int action);
+static bool postmaster_is_alive(pid_t pid);
+static void wait_postmaster_connection(const char *conninfo);
+static void wait_for_end_recovery(const char *conninfo);
+static void create_publication(PGconn *conn, LogicalRepInfo *dbinfo);
+static void drop_publication(PGconn *conn, LogicalRepInfo *dbinfo);
+static void create_subscription(PGconn *conn, LogicalRepInfo *dbinfo);
+static void drop_subscription(PGconn *conn, LogicalRepInfo *dbinfo);
+static void set_replication_progress(PGconn *conn, LogicalRepInfo *dbinfo, const char *lsn);
+static void enable_subscription(PGconn *conn, LogicalRepInfo *dbinfo);
+
+#define USEC_PER_SEC 1000000
+#define WAIT_INTERVAL 1 /* 1 second */
+
+/* Options */
+const char *progname;
+static char *subscriber_dir = NULL;
+static char *pub_conninfo_str = NULL;
+static char *sub_conninfo_str = NULL;
+static SimpleStringList database_names = {NULL, NULL};
+static int verbose = 0;
+static bool success = false;
+
+static LogicalRepInfo *dbinfo;
+
+static int num_dbs = 0;
+
+static char temp_replslot[NAMEDATALEN];
+static bool made_temp_replslot = false;
+
+char pidfile[MAXPGPATH]; /* subscriber PID file */
+
+enum WaitPMResult
+{
+ POSTMASTER_READY,
+ POSTMASTER_STANDBY,
+ POSTMASTER_STILL_STARTING,
+ POSTMASTER_FAILED
+};
+
+
+/*
+ * Cleanup objects that were created by pg_subscriber if there is an error.
+ *
+ * Replication slots, publications and subscriptions are created. Dependind on
+ * the step it failed, it should remove the already created objects if it is
+ * possible (sometimes it won't work due to a connection issue).
+ */
+static void
+cleanup_objects_atexit(void)
+{
+ PGconn *conn;
+ int i;
+
+ if (success)
+ return;
+
+ for (i = 0; i < num_dbs; i++)
+ {
+ if (dbinfo[i].made_subscription)
+ {
+ conn = connect_database(dbinfo[i].subconninfo, true);
+ if (conn != NULL)
+ {
+ drop_subscription(conn, &dbinfo[i]);
+ disconnect_database(conn);
+ }
+ }
+
+ if (dbinfo[i].made_publication || dbinfo[i].made_replslot)
+ {
+ conn = connect_database(dbinfo[i].pubconninfo, true);
+ if (conn != NULL)
+ {
+ if (dbinfo[i].made_publication)
+ drop_publication(conn, &dbinfo[i]);
+ if (dbinfo[i].made_replslot)
+ drop_replication_slot(conn, &dbinfo[i], NULL);
+ disconnect_database(conn);
+ }
+ }
+ }
+
+ if (made_temp_replslot)
+ {
+ conn = connect_database(dbinfo[0].pubconninfo, true);
+ drop_replication_slot(conn, &dbinfo[0], temp_replslot);
+ disconnect_database(conn);
+ }
+}
+
+static void
+usage(void)
+{
+ printf(_("%s creates a new logical replica from a base backup or a standby server.\n\n"),
+ progname);
+ printf(_("Usage:\n"));
+ printf(_(" %s [OPTION]...\n"), progname);
+ printf(_("\nOptions:\n"));
+ printf(_(" -D, --pgdata=DATADIR location for the subscriber data directory\n"));
+ printf(_(" -P, --publisher-conninfo=CONNINFO publisher connection string\n"));
+ printf(_(" -S, --subscriber-conninfo=CONNINFO subscriber connection string\n"));
+ printf(_(" -d, --database=DBNAME database to create a subscription\n"));
+ printf(_(" -v, --verbose output verbose messages\n"));
+ printf(_(" -V, --version output version information, then exit\n"));
+ printf(_(" -?, --help show this help, then exit\n"));
+ printf(_("\nReport bugs to <%s>.\n"), PACKAGE_BUGREPORT);
+ printf(_("%s home page: <%s>\n"), PACKAGE_NAME, PACKAGE_URL);
+}
+
+/*
+ * Validate a connection string. Returns a base connection string that is a
+ * connection string without a database name plus a fallback application name.
+ * Since we might process multiple databases, each database name will be
+ * appended to this base connection string to provide a final connection string.
+ * If the second argument (dbname) is not null, returns dbname if the provided
+ * connection string contains it. If option --database is not provided, uses
+ * dbname as the only database to setup the logical replica.
+ * It is the caller's responsibility to free the returned connection string and
+ * dbname.
+ */
+static char *
+get_base_conninfo(char *conninfo, char *dbname, const char *noderole)
+{
+ PQExpBuffer buf = createPQExpBuffer();
+ PQconninfoOption *conn_opts = NULL;
+ PQconninfoOption *conn_opt;
+ char *errmsg = NULL;
+ char *ret;
+ int i;
+
+ if (verbose)
+ pg_log_info("validating connection string on %s", noderole);
+
+ conn_opts = PQconninfoParse(conninfo, &errmsg);
+ if (conn_opts == NULL)
+ {
+ pg_log_error("could not parse connection string: %s", errmsg);
+ return NULL;
+ }
+
+ i = 0;
+ for (conn_opt = conn_opts; conn_opt->keyword != NULL; conn_opt++)
+ {
+ if (strcmp(conn_opt->keyword, "dbname") == 0 && conn_opt->val != NULL)
+ {
+ if (dbname)
+ dbname = pg_strdup(conn_opt->val);
+ continue;
+ }
+
+ if (conn_opt->val != NULL && conn_opt->val[0] != '\0')
+ {
+ if (i > 0)
+ appendPQExpBufferChar(buf, ' ');
+ appendPQExpBuffer(buf, "%s=%s", conn_opt->keyword, conn_opt->val);
+ i++;
+ }
+ }
+
+ if (i > 0)
+ appendPQExpBufferChar(buf, ' ');
+ appendPQExpBuffer(buf, "fallback_application_name=%s", progname);
+
+ ret = pg_strdup(buf->data);
+
+ destroyPQExpBuffer(buf);
+ PQconninfoFree(conn_opts);
+
+ return ret;
+}
+
+/*
+ * Is it a cluster directory? These are preliminary checks. It is far from
+ * making an accurate check. If it is not a clone from the publisher, it will
+ * eventually fail in a future step.
+ */
+static bool
+check_data_directory(const char *datadir)
+{
+ struct stat statbuf;
+ char versionfile[MAXPGPATH];
+
+ if (verbose)
+ pg_log_info("checking if directory \"%s\" is a cluster data directory",
+ datadir);
+
+ if (stat(datadir, &statbuf) != 0)
+ {
+ if (errno == ENOENT)
+ pg_log_error("data directory \"%s\" does not exist", datadir);
+ else
+ pg_log_error("could not access directory \"%s\": %s", datadir, strerror(errno));
+
+ return false;
+ }
+
+ snprintf(versionfile, MAXPGPATH, "%s/PG_VERSION", datadir);
+ if (stat(versionfile, &statbuf) != 0 && errno == ENOENT)
+ {
+ pg_log_error("directory \"%s\" is not a database cluster directory", datadir);
+ return false;
+ }
+
+ return true;
+}
+
+/*
+ * Append database name into a base connection string.
+ *
+ * dbname is the only parameter that changes so it is not included in the base
+ * connection string. This function concatenates dbname to build a "real"
+ * connection string.
+ */
+static char *
+concat_conninfo_dbname(const char *conninfo, const char *dbname)
+{
+ PQExpBuffer buf = createPQExpBuffer();
+ char *ret;
+
+ Assert(conninfo != NULL);
+
+ appendPQExpBufferStr(buf, conninfo);
+ appendPQExpBuffer(buf, " dbname=%s", dbname);
+ appendPQExpBufferStr(buf, " replication=database");
+
+ ret = pg_strdup(buf->data);
+ destroyPQExpBuffer(buf);
+
+ return ret;
+}
+
+static PGconn *
+connect_database(const char *conninfo, bool secure_search_path)
+{
+ PGconn *conn;
+
+ conn = PQconnectdb(conninfo);
+ if (PQstatus(conn) != CONNECTION_OK)
+ {
+ pg_log_error("connection to database failed: %s", PQerrorMessage(conn));
+ return NULL;
+ }
+
+ /* secure search_path */
+ if (secure_search_path)
+ {
+ PGresult *res;
+
+ res = PQexec(conn, ALWAYS_SECURE_SEARCH_PATH_SQL);
+ if (PQresultStatus(res) != PGRES_TUPLES_OK)
+ {
+ pg_log_error("could not clear search_path: %s", PQresultErrorMessage(res));
+ return NULL;
+ }
+ PQclear(res);
+ }
+
+ return conn;
+}
+
+static void
+disconnect_database(PGconn *conn)
+{
+ Assert(conn != NULL);
+
+ PQfinish(conn);
+}
+
+/*
+ * Obtain the system identifier using the provided connection. It will be used
+ * to compare if a data directory is a clone of another one.
+ */
+static char *
+get_sysid_from_conn(const char *conninfo)
+{
+ PGconn *conn;
+ PGresult *res;
+ char *repconninfo;
+ char *sysid = NULL;
+
+ if (verbose)
+ pg_log_info("getting system identifier from publisher");
+
+ repconninfo = psprintf("%s replication=database", conninfo);
+ conn = connect_database(repconninfo, false);
+ if (conn == NULL)
+ exit(1);
+
+ res = PQexec(conn, "IDENTIFY_SYSTEM");
+ if (PQresultStatus(res) != PGRES_TUPLES_OK)
+ {
+ pg_log_error("could not send replication command \"%s\": %s",
+ "IDENTIFY_SYSTEM", PQresultErrorMessage(res));
+ PQclear(res);
+ return NULL;
+ }
+ if (PQntuples(res) != 1 || PQnfields(res) < 3)
+ {
+ pg_log_error("could not identify system: got %d rows and %d fields, expected %d rows and %d or more fields",
+ PQntuples(res), PQnfields(res), 1, 3);
+
+ PQclear(res);
+ return NULL;
+ }
+
+ sysid = pg_strdup(PQgetvalue(res, 0, 0));
+
+ disconnect_database(conn);
+
+ return sysid;
+}
+
+/*
+ * Obtain the system identifier from control file. It will be used to compare
+ * if a data directory is a clone of another one. This routine is used locally
+ * and avoids a replication connection.
+ */
+static char *
+get_control_from_datadir(const char *datadir)
+{
+ ControlFileData *cf;
+ bool crc_ok;
+ char *sysid = pg_malloc(32);
+
+ if (verbose)
+ pg_log_info("getting system identifier from subscriber");
+
+ cf = get_controlfile(datadir, &crc_ok);
+ if (!crc_ok)
+ {
+ pg_log_error("control file appears to be corrupt");
+ exit(1);
+ }
+
+ snprintf(sysid, 32, UINT64_FORMAT, cf->system_identifier);
+
+ pfree(cf);
+
+ return sysid;
+}
+
+/*
+ * Create a logical replication slot and returns a consistent LSN. The returned
+ * LSN might be used to catch up the subscriber up to the required point.
+ *
+ * XXX CreateReplicationSlot() is not used because it does not provide the one-row
+ * result set that contains the consistent LSN.
+ */
+static char *
+create_logical_replication_slot(PGconn *conn, LogicalRepInfo *dbinfo,
+ const char *slot_name)
+{
+ PQExpBuffer str = createPQExpBuffer();
+ PGresult *res;
+ char *lsn = NULL;
+
+ Assert(conn != NULL);
+
+ if (verbose)
+ pg_log_info("creating the replication slot \"%s\" on database \"%s\"", slot_name, dbinfo->dbname);
+
+ appendPQExpBuffer(str, "CREATE_REPLICATION_SLOT \"%s\"", slot_name);
+ appendPQExpBufferStr(str, " LOGICAL \"pgoutput\" NOEXPORT_SNAPSHOT");
+
+ if (verbose)
+ pg_log_info("command is: %s", str->data);
+
+ res = PQexec(conn, str->data);
+ if (PQresultStatus(res) != PGRES_TUPLES_OK)
+ {
+ pg_log_error("could not create replication slot \"%s\" on database \"%s\": %s", slot_name, dbinfo->dbname,
+ PQresultErrorMessage(res));
+ return lsn;
+ }
+
+ /* for cleanup purposes */
+ if (slot_name == NULL)
+ dbinfo->made_replslot = true;
+ else
+ made_temp_replslot = true;
+
+ lsn = pg_strdup(PQgetvalue(res, 0, 1));
+
+ PQclear(res);
+ destroyPQExpBuffer(str);
+
+ return lsn;
+}
+
+static void
+drop_replication_slot(PGconn *conn, LogicalRepInfo *dbinfo, const char *slot_name)
+{
+ PQExpBuffer str = createPQExpBuffer();
+ PGresult *res;
+
+ Assert(conn != NULL);
+
+ if (verbose)
+ pg_log_info("dropping the replication slot \"%s\" on database \"%s\"", slot_name, dbinfo->dbname);
+
+ appendPQExpBuffer(str, "DROP_REPLICATION_SLOT \"%s\"", slot_name);
+
+ res = PQexec(conn, str->data);
+ if (PQresultStatus(res) != PGRES_COMMAND_OK)
+ pg_log_error("could not drop replication slot \"%s\" on database \"%s\": %s", slot_name, dbinfo->dbname,
+ PQerrorMessage(conn));
+
+ PQclear(res);
+ destroyPQExpBuffer(str);
+}
+
+/*
+ * Reports a suitable message if pg_ctl fails.
+ */
+static void
+pg_ctl_status(const char *pg_ctl_cmd, int rc, int action)
+{
+ if (rc != 0)
+ {
+ if (WIFEXITED(rc))
+ {
+ pg_log_error("pg_ctl failed with exit code %d", WEXITSTATUS(rc));
+ }
+ else if (WIFSIGNALED(rc))
+ {
+#if defined(WIN32)
+ pg_log_error("pg_ctl was terminated by exception 0x%X", WTERMSIG(rc));
+ fprintf(stderr,
+ "See C include file \"ntstatus.h\" for a description of the hexadecimal value.\n");
+#else
+ pg_log_error("pg_ctl was terminated by signal %d: %s",
+ WTERMSIG(rc), pg_strsignal(WTERMSIG(rc)));
+#endif
+ }
+ else
+ {
+ pg_log_error("pg_ctl exited with unrecognized status %d", rc);
+ }
+
+ fprintf(stderr, "The failed command was: %s\n", pg_ctl_cmd);
+ exit(1);
+ }
+
+ if (verbose)
+ {
+ if (action)
+ pg_log_info("postmaster was started");
+ else
+ pg_log_info("postmaster was stopped");
+ }
+}
+
+/*
+ * XXX This function was copied from pg_ctl.c.
+ *
+ * We should probably move it to a common place.
+ */
+static bool
+postmaster_is_alive(pid_t pid)
+{
+ /*
+ * Test to see if the process is still there. Note that we do not
+ * consider an EPERM failure to mean that the process is still there;
+ * EPERM must mean that the given PID belongs to some other userid, and
+ * considering the permissions on $PGDATA, that means it's not the
+ * postmaster we are after.
+ *
+ * Don't believe that our own PID or parent shell's PID is the postmaster,
+ * either. (Windows hasn't got getppid(), though.)
+ */
+ if (pid == getpid())
+ return false;
+#ifndef WIN32
+ if (pid == getppid())
+ return false;
+#endif
+ if (kill(pid, 0) == 0)
+ return true;
+ return false;
+}
+
+/*
+ * Returns after postmaster is accepting connections.
+ */
+static void
+wait_postmaster_connection(const char *conninfo)
+{
+ PGPing ret;
+ long pmpid;
+ int status = POSTMASTER_STILL_STARTING;
+
+ if (verbose)
+ pg_log_info("waiting for the postmaster to allow connections ...");
+
+ /*
+ * Wait postmaster to come up. XXX this code path is a modified version of
+ * wait_for_postmaster().
+ */
+ for (;;)
+ {
+ char **optlines;
+ int numlines;
+
+ if ((optlines = readfile(pidfile, &numlines)) != NULL &&
+ numlines >= LOCK_FILE_LINE_PM_STATUS)
+ {
+ /*
+ * Check the status line (this assumes a v10 or later server).
+ */
+ char *pmstatus = optlines[LOCK_FILE_LINE_PM_STATUS - 1];
+
+ pmpid = atol(optlines[LOCK_FILE_LINE_PID - 1]);
+
+ if (strcmp(pmstatus, PM_STATUS_READY) == 0)
+ {
+ free_readfile(optlines);
+ status = POSTMASTER_READY;
+ break;
+ }
+ else if (strcmp(pmstatus, PM_STATUS_STANDBY) == 0)
+ {
+ free_readfile(optlines);
+ status = POSTMASTER_STANDBY;
+ break;
+ }
+ }
+
+ free_readfile(optlines);
+
+ pg_usleep(WAIT_INTERVAL * USEC_PER_SEC);
+ }
+
+ if (verbose)
+ pg_log_info("postmaster.pid is available");
+
+ if (status == POSTMASTER_STILL_STARTING)
+ {
+ pg_log_error("server did not start in time");
+ exit(1);
+ }
+ else if (status == POSTMASTER_STANDBY)
+ {
+ pg_log_error("server is running but hot standby mode is not enabled");
+ exit(1);
+ }
+ else if (status == POSTMASTER_FAILED)
+ {
+ pg_log_error("could not start server");
+ fprintf(stderr, "Examine the log output.\n");
+ exit(1);
+ }
+
+ if (verbose)
+ {
+ pg_log_info("postmaster is up and running");
+ pg_log_info("waiting until the postmaster accepts connections ...");
+ }
+
+ /* Postmaster is up. Let's wait for it to accept connections. */
+ for (;;)
+ {
+ ret = PQping(conninfo);
+ if (ret == PQPING_OK)
+ break;
+ else if (ret == PQPING_NO_ATTEMPT)
+ break;
+
+ /*
+ * Postmaster started but for some reason it crashed leaving a
+ * postmaster.pid.
+ */
+ if (!postmaster_is_alive((pid_t) pmpid))
+ {
+ pg_log_error("could not start server");
+ fprintf(stderr, "Examine the log output.\n");
+ exit(1);
+ }
+
+ pg_usleep(WAIT_INTERVAL * USEC_PER_SEC);
+ }
+
+ if (verbose)
+ pg_log_info("postmaster is accepting connections");
+}
+
+/*
+ * Returns after the server finishes the recovery process.
+ */
+static void
+wait_for_end_recovery(const char *conninfo)
+{
+ PGconn *conn;
+ PGresult *res;
+ int status = POSTMASTER_STILL_STARTING;
+
+ if (verbose)
+ pg_log_info("waiting the postmaster to reach the consistent state ...");
+
+ conn = connect_database(conninfo, true);
+ if (conn == NULL)
+ exit(1);
+
+ for (;;)
+ {
+ bool in_recovery;
+
+ res = PQexec(conn, "SELECT pg_catalog.pg_is_in_recovery()");
+
+ if (PQresultStatus(res) != PGRES_TUPLES_OK)
+ {
+ pg_log_error("could not obtain recovery progress");
+ exit(1);
+ }
+
+ if (PQntuples(res) != 1)
+ {
+ pg_log_error("unexpected result from pg_is_in_recovery function");
+ exit(1);
+ }
+
+ in_recovery = (strcmp(PQgetvalue(res, 0, 0), "t") == 0);
+
+ PQclear(res);
+
+ /* Does the recovery process finish? */
+ if (!in_recovery)
+ {
+ status = POSTMASTER_READY;
+ break;
+ }
+
+ /* Keep waiting. */
+ pg_usleep(WAIT_INTERVAL * USEC_PER_SEC);
+ }
+
+ disconnect_database(conn);
+
+ if (status == POSTMASTER_STILL_STARTING)
+ {
+ pg_log_error("server did not end recovery");
+ exit(1);
+ }
+
+ if (verbose)
+ pg_log_info("postmaster reached the consistent state");
+}
+
+/*
+ * Create a publication that includes all tables in the database.
+ */
+static void
+create_publication(PGconn *conn, LogicalRepInfo *dbinfo)
+{
+ PQExpBuffer str = createPQExpBuffer();
+ PGresult *res;
+
+ Assert(conn != NULL);
+
+ /* Check if the publication needs to be created. */
+ appendPQExpBuffer(str,
+ "SELECT puballtables FROM pg_catalog.pg_publication WHERE pubname = '%s'",
+ dbinfo->pubname);
+ res = PQexec(conn, str->data);
+ if (PQresultStatus(res) != PGRES_TUPLES_OK)
+ {
+ pg_log_error("could not obtain publication information: %s",
+ PQresultErrorMessage(res));
+ PQclear(res);
+ PQfinish(conn);
+ exit(1);
+ }
+
+ if (PQntuples(res) == 1)
+ {
+ /*
+ * If publication name already exists and puballtables is true, let's
+ * use it. A previous run of pg_subscriber must have created this
+ * publication. Bail out.
+ */
+ if (strcmp(PQgetvalue(res, 0, 0), "t") == 0)
+ {
+ if (verbose)
+ pg_log_info("publication \"%s\" already exists", dbinfo->pubname);
+ return;
+ }
+ else
+ {
+ /*
+ * XXX Unfortunately, if it reaches this code path, pg_subscriber
+ * will always fail here. That's bad but it is not expected that
+ * the user choose a name with pg_subscriber_ prefix followed by
+ * the exact database oid in which puballtables is false.
+ */
+ pg_log_error("publication \"%s\" does not replicate changes for all tables",
+ dbinfo->pubname);
+ PQclear(res);
+ PQfinish(conn);
+ exit(1);
+ }
+ }
+
+ PQclear(res);
+ resetPQExpBuffer(str);
+
+ if (verbose)
+ pg_log_info("creating publication \"%s\" on database \"%s\"", dbinfo->pubname, dbinfo->dbname);
+
+ appendPQExpBuffer(str, "CREATE PUBLICATION %s FOR ALL TABLES", dbinfo->pubname);
+
+ if (verbose)
+ pg_log_info("command is: %s", str->data);
+
+ res = PQexec(conn, str->data);
+ if (PQresultStatus(res) != PGRES_COMMAND_OK)
+ {
+ pg_log_error("could not create publication \"%s\" on database \"%s\": %s",
+ dbinfo->pubname, dbinfo->dbname, PQerrorMessage(conn));
+ PQfinish(conn);
+ exit(1);
+ }
+
+ /* for cleanup purposes */
+ dbinfo->made_publication = true;
+
+ PQclear(res);
+ destroyPQExpBuffer(str);
+}
+
+/*
+ * Remove publication if it couldn't finish all steps.
+ */
+static void
+drop_publication(PGconn *conn, LogicalRepInfo *dbinfo)
+{
+ PQExpBuffer str = createPQExpBuffer();
+ PGresult *res;
+
+ Assert(conn != NULL);
+
+ if (verbose)
+ pg_log_info("dropping publication \"%s\" on database \"%s\"", dbinfo->pubname, dbinfo->dbname);
+
+ appendPQExpBuffer(str, "DROP PUBLICATION %s", dbinfo->pubname);
+
+ if (verbose)
+ pg_log_info("command is: %s", str->data);
+
+ res = PQexec(conn, str->data);
+ if (PQresultStatus(res) != PGRES_COMMAND_OK)
+ pg_log_error("could not drop publication \"%s\" on database \"%s\": %s", dbinfo->pubname, dbinfo->dbname, PQerrorMessage(conn));
+
+ PQclear(res);
+ destroyPQExpBuffer(str);
+}
+
+/*
+ * Create a subscription with some predefined options.
+ *
+ * A replication slot was already created in a previous step. Let's use it. By
+ * default, the subscription name is used as replication slot name. It is
+ * not required to copy data. The subscription will be created but it will not
+ * be enabled now. That's because the replication progress must be set and the
+ * replication origin name (one of the function arguments) contains the
+ * subscription OID in its name. Once the subscription is created,
+ * set_replication_progress() can obtain the chosen origin name and set up its
+ * initial location.
+ */
+static void
+create_subscription(PGconn *conn, LogicalRepInfo *dbinfo)
+{
+ PQExpBuffer str = createPQExpBuffer();
+ PGresult *res;
+
+ Assert(conn != NULL);
+
+ if (verbose)
+ pg_log_info("creating subscription \"%s\" on database \"%s\"", dbinfo->subname, dbinfo->dbname);
+
+ appendPQExpBuffer(str,
+ "CREATE SUBSCRIPTION %s CONNECTION '%s' PUBLICATION %s "
+ "WITH (create_slot = false, copy_data = false, enabled = false)",
+ dbinfo->subname, dbinfo->pubconninfo, dbinfo->pubname);
+
+ if (verbose)
+ pg_log_info("command is: %s", str->data);
+
+ res = PQexec(conn, str->data);
+ if (PQresultStatus(res) != PGRES_COMMAND_OK)
+ {
+ pg_log_error("could not create subscription \"%s\" on database \"%s\": %s",
+ dbinfo->subname, dbinfo->dbname, PQerrorMessage(conn));
+ PQfinish(conn);
+ exit(1);
+ }
+
+ /* for cleanup purposes */
+ dbinfo->made_subscription = true;
+
+ PQclear(res);
+ destroyPQExpBuffer(str);
+}
+
+/*
+ * Remove subscription if it couldn't finish all steps.
+ */
+static void
+drop_subscription(PGconn *conn, LogicalRepInfo *dbinfo)
+{
+ PQExpBuffer str = createPQExpBuffer();
+ PGresult *res;
+
+ Assert(conn != NULL);
+
+ if (verbose)
+ pg_log_info("dropping subscription \"%s\" on database \"%s\"", dbinfo->subname, dbinfo->dbname);
+
+ appendPQExpBuffer(str, "DROP SUBSCRIPTION %s", dbinfo->subname);
+
+ if (verbose)
+ pg_log_info("command is: %s", str->data);
+
+ res = PQexec(conn, str->data);
+ if (PQresultStatus(res) != PGRES_COMMAND_OK)
+ pg_log_error("could not drop subscription \"%s\" on database \"%s\": %s", dbinfo->subname, dbinfo->dbname, PQerrorMessage(conn));
+
+ PQclear(res);
+ destroyPQExpBuffer(str);
+}
+
+/*
+ * Sets the replication progress to the consistent LSN.
+ *
+ * The subscriber caught up to the consistent LSN provided by the temporary
+ * replication slot. The goal is to set up the initial location for the logical
+ * replication that is the exact LSN that the subscriber was promoted. Once the
+ * subscription is enabled it will start streaming from that location onwards.
+ */
+static void
+set_replication_progress(PGconn *conn, LogicalRepInfo *dbinfo, const char *lsn)
+{
+ PQExpBuffer str = createPQExpBuffer();
+ PGresult *res;
+ Oid suboid;
+ char originname[NAMEDATALEN];
+
+ Assert(conn != NULL);
+
+ appendPQExpBuffer(str,
+ "SELECT oid FROM pg_catalog.pg_subscription WHERE subname = '%s'", dbinfo->subname);
+
+ res = PQexec(conn, str->data);
+ if (PQresultStatus(res) != PGRES_TUPLES_OK)
+ {
+ pg_log_error("could not obtain subscription OID: %s",
+ PQresultErrorMessage(res));
+ PQclear(res);
+ PQfinish(conn);
+ exit(1);
+ }
+
+ if (PQntuples(res) != 1)
+ {
+ pg_log_error("could not obtain subscription OID: got %d rows, expected %d rows",
+ PQntuples(res), 1);
+ PQclear(res);
+ PQfinish(conn);
+ exit(1);
+ }
+
+ /*
+ * The origin name is defined as pg_%u. %u is the subscription OID. See
+ * ApplyWorkerMain().
+ */
+ suboid = strtoul(PQgetvalue(res, 0, 0), NULL, 10);
+ snprintf(originname, sizeof(originname), "pg_%u", suboid);
+
+ PQclear(res);
+
+ if (verbose)
+ pg_log_info("setting the replication progress (node name \"%s\" ; LSN %s) on database \"%s\"",
+ originname, lsn, dbinfo->dbname);
+
+ resetPQExpBuffer(str);
+ appendPQExpBuffer(str,
+ "SELECT pg_catalog.pg_replication_origin_advance('%s', '%s')", originname, lsn);
+
+ if (verbose)
+ pg_log_info("command is: %s", str->data);
+
+ res = PQexec(conn, str->data);
+ if (PQresultStatus(res) != PGRES_TUPLES_OK)
+ {
+ pg_log_error("could not set replication progress for the subscription \"%s\": %s",
+ dbinfo->subname, PQresultErrorMessage(res));
+ PQfinish(conn);
+ exit(1);
+ }
+
+ PQclear(res);
+ destroyPQExpBuffer(str);
+}
+
+/*
+ * Enables the subscription.
+ *
+ * The subscription was created in a previous step but it was disabled. After
+ * adjusting the initial location, enabling the subscription is the last step
+ * of this setup.
+ */
+static void
+enable_subscription(PGconn *conn, LogicalRepInfo *dbinfo)
+{
+ PQExpBuffer str = createPQExpBuffer();
+ PGresult *res;
+
+ Assert(conn != NULL);
+
+ if (verbose)
+ pg_log_info("enabling subscription \"%s\" on database \"%s\"", dbinfo->subname, dbinfo->dbname);
+
+ appendPQExpBuffer(str, "ALTER SUBSCRIPTION %s ENABLE", dbinfo->subname);
+
+ if (verbose)
+ pg_log_info("command is: %s", str->data);
+
+ res = PQexec(conn, str->data);
+ if (PQresultStatus(res) != PGRES_COMMAND_OK)
+ {
+ pg_log_error("could not enable subscription \"%s\": %s", dbinfo->subname,
+ PQerrorMessage(conn));
+ PQfinish(conn);
+ exit(1);
+ }
+
+ PQclear(res);
+ destroyPQExpBuffer(str);
+}
+
+int
+main(int argc, char **argv)
+{
+ static struct option long_options[] =
+ {
+ {"help", no_argument, NULL, '?'},
+ {"version", no_argument, NULL, 'V'},
+ {"pgdata", required_argument, NULL, 'D'},
+ {"publisher-conninfo", required_argument, NULL, 'P'},
+ {"subscriber-conninfo", required_argument, NULL, 'S'},
+ {"database", required_argument, NULL, 'd'},
+ {"verbose", no_argument, NULL, 'v'},
+ {"stop-subscriber", no_argument, NULL, 1},
+ {NULL, 0, NULL, 0}
+ };
+
+ int c;
+ int option_index;
+
+ char *pg_ctl_path;
+ char *pg_ctl_cmd;
+ int rc;
+
+ SimpleStringListCell *cell;
+
+ char *pub_base_conninfo = NULL;
+ char *sub_base_conninfo = NULL;
+ char *dbname_conninfo;
+
+ char *pub_sysid;
+ char *sub_sysid;
+ struct stat statbuf;
+
+ PGconn *conn;
+ char *consistent_lsn;
+
+ PQExpBuffer recoveryconfcontents = NULL;
+
+ int i;
+
+ pg_logging_init(argv[0]);
+ progname = get_progname(argv[0]);
+ set_pglocale_pgservice(argv[0], PG_TEXTDOMAIN("pg_subscriber"));
+
+ if (argc > 1)
+ {
+ if (strcmp(argv[1], "--help") == 0 || strcmp(argv[1], "-?") == 0)
+ {
+ usage();
+ exit(0);
+ }
+ else if (strcmp(argv[1], "-V") == 0
+ || strcmp(argv[1], "--version") == 0)
+ {
+ puts("pg_subscriber (PostgreSQL) " PG_VERSION);
+ exit(0);
+ }
+ }
+
+ atexit(cleanup_objects_atexit);
+
+ /*
+ * Don't allow it to be run as root. It uses pg_ctl which does not allow
+ * it either.
+ */
+#ifndef WIN32
+ if (geteuid() == 0)
+ {
+ pg_log_error("cannot be executed by \"root\"");
+ fprintf(stderr, _("You must run %s as the PostgreSQL superuser.\n"),
+ progname);
+ exit(1);
+ }
+#endif
+
+ while ((c = getopt_long(argc, argv, "D:P:S:d:t:v",
+ long_options, &option_index)) != -1)
+ {
+ switch (c)
+ {
+ case 'D':
+ subscriber_dir = pg_strdup(optarg);
+ break;
+ case 'P':
+ pub_conninfo_str = pg_strdup(optarg);
+ break;
+ case 'S':
+ sub_conninfo_str = pg_strdup(optarg);
+ break;
+ case 'd':
+ simple_string_list_append(&database_names, optarg);
+ num_dbs++;
+ break;
+ case 'v':
+ verbose++;
+ break;
+ default:
+
+ /*
+ * getopt_long already emitted a complaint
+ */
+ fprintf(stderr, _("Try \"%s --help\" for more information.\n"),
+ progname);
+ exit(1);
+ }
+ }
+
+ /*
+ * Any non-option arguments?
+ */
+ if (optind < argc)
+ {
+ pg_log_error("too many command-line arguments (first is \"%s\")",
+ argv[optind]);
+ fprintf(stderr, _("Try \"%s --help\" for more information.\n"),
+ progname);
+ exit(1);
+ }
+
+ /*
+ * Required arguments
+ */
+ if (subscriber_dir == NULL)
+ {
+ pg_log_error("no subscriber data directory specified");
+ fprintf(stderr, _("Try \"%s --help\" for more information.\n"),
+ progname);
+ exit(1);
+ }
+
+ /*
+ * Parse connection string. Build a base connection string that might be
+ * reused by multiple databases.
+ */
+ if (pub_conninfo_str == NULL)
+ {
+ /*
+ * FIXME use primary_conninfo (if available) from subscriber and
+ * extract publisher connection string. Assume that there are
+ * identical entries for physical and logical replication. If there is
+ * not, we would fail anyway.
+ */
+ pg_log_error("no publisher connection string specified");
+ fprintf(stderr, _("Try \"%s --help\" for more information.\n"),
+ progname);
+ exit(1);
+ }
+ dbname_conninfo = pg_malloc(NAMEDATALEN);
+ pub_base_conninfo = get_base_conninfo(pub_conninfo_str, dbname_conninfo,
+ "publisher");
+ if (pub_base_conninfo == NULL)
+ exit(1);
+
+ if (sub_conninfo_str == NULL)
+ {
+ pg_log_error("no subscriber connection string specified");
+ fprintf(stderr, _("Try \"%s --help\" for more information.\n"),
+ progname);
+ exit(1);
+ }
+ sub_base_conninfo = get_base_conninfo(sub_conninfo_str, NULL, "subscriber");
+ if (sub_base_conninfo == NULL)
+ exit(1);
+
+ if (database_names.head == NULL)
+ {
+ if (verbose)
+ pg_log_info("no database was specified");
+
+ /*
+ * If --database option is not provided, try to obtain the dbname from
+ * the publisher conninfo. If dbname parameter is not available, error
+ * out.
+ */
+ if (dbname_conninfo)
+ {
+ simple_string_list_append(&database_names, dbname_conninfo);
+ num_dbs++;
+
+ if (verbose)
+ pg_log_info("database \"%s\" was extracted from the publisher connection string",
+ dbname_conninfo);
+ }
+ else
+ {
+ pg_log_error("no database name specified");
+ fprintf(stderr, _("Try \"%s --help\" for more information.\n"),
+ progname);
+ exit(1);
+ }
+ }
+
+ /*
+ * Get the absolute pg_ctl path on the subscriber.
+ */
+ pg_ctl_path = pg_malloc(MAXPGPATH);
+ rc = find_other_exec(argv[0], "pg_ctl",
+ "pg_ctl (PostgreSQL) " PG_VERSION "\n",
+ pg_ctl_path);
+ if (rc < 0)
+ {
+ char full_path[MAXPGPATH];
+
+ if (find_my_exec(argv[0], full_path) < 0)
+ strlcpy(full_path, progname, sizeof(full_path));
+ if (rc == -1)
+ pg_log_error("The program \"%s\" is needed by %s but was not found in the\n"
+ "same directory as \"%s\".\n"
+ "Check your installation.",
+ "pg_ctl", progname, full_path);
+ else
+ pg_log_error("The program \"%s\" was found by \"%s\"\n"
+ "but was not the same version as %s.\n"
+ "Check your installation.",
+ "pg_ctl", full_path, progname);
+ exit(1);
+ }
+
+ if (verbose)
+ pg_log_info("pg_ctl path is: %s", pg_ctl_path);
+
+ /* rudimentary check for a data directory. */
+ if (!check_data_directory(subscriber_dir))
+ exit(1);
+
+ /* subscriber PID file. */
+ snprintf(pidfile, MAXPGPATH, "%s/postmaster.pid", subscriber_dir);
+
+ /* Store database information for publisher and subscriber. */
+ dbinfo = (LogicalRepInfo *) pg_malloc(num_dbs * sizeof(LogicalRepInfo));
+ i = 0;
+ for (cell = database_names.head; cell; cell = cell->next)
+ {
+ char *conninfo;
+
+ /* Publisher. */
+ conninfo = concat_conninfo_dbname(pub_base_conninfo, cell->val);
+ dbinfo[i].pubconninfo = conninfo;
+ dbinfo[i].dbname = cell->val;
+ dbinfo[i].made_replslot = false;
+ dbinfo[i].made_publication = false;
+ dbinfo[i].made_subscription = false;
+ /* other struct fields will be filled later. */
+
+ /* Subscriber. */
+ conninfo = concat_conninfo_dbname(sub_base_conninfo, cell->val);
+ dbinfo[i].subconninfo = conninfo;
+
+ i++;
+ }
+
+ /*
+ * Check if the subscriber data directory has the same system identifier
+ * than the publisher data directory.
+ */
+ pub_sysid = pg_malloc(32);
+ pub_sysid = get_sysid_from_conn(dbinfo[0].pubconninfo);
+ sub_sysid = pg_malloc(32);
+ sub_sysid = get_control_from_datadir(subscriber_dir);
+ if (strcmp(pub_sysid, sub_sysid) != 0)
+ {
+ pg_log_error("subscriber data directory is not a base backup from the publisher");
+ exit(1);
+ }
+
+ /*
+ * Stop the subscriber if it is a standby server. Before executing the
+ * transformation steps, make sure the subscriber is not running because
+ * one of the steps is to modify some recovery parameters that require a
+ * restart.
+ */
+ if (stat(pidfile, &statbuf) == 0)
+ {
+ if (verbose)
+ {
+ pg_log_info("subscriber is up and running");
+ pg_log_info("stopping the server to start the transformation steps");
+ }
+
+ pg_ctl_cmd = psprintf("\"%s\" stop -D \"%s\" -s", pg_ctl_path, subscriber_dir);
+ rc = system(pg_ctl_cmd);
+ pg_ctl_status(pg_ctl_cmd, rc, 0);
+ }
+
+ /*
+ * Create a replication slot for each database on the publisher.
+ */
+ for (i = 0; i < num_dbs; i++)
+ {
+ PGresult *res;
+ char replslotname[NAMEDATALEN];
+
+ conn = connect_database(dbinfo[i].pubconninfo, true);
+ if (conn == NULL)
+ exit(1);
+
+ res = PQexec(conn,
+ "SELECT oid FROM pg_catalog.pg_database WHERE datname = current_database()");
+ if (PQresultStatus(res) != PGRES_TUPLES_OK)
+ {
+ pg_log_error("could not obtain database OID: %s", PQresultErrorMessage(res));
+ PQclear(res);
+ PQfinish(conn);
+ exit(1);
+ }
+
+ if (PQntuples(res) != 1)
+ {
+ pg_log_error("could not obtain database OID: got %d rows, expected %d rows",
+ PQntuples(res), 1);
+ PQclear(res);
+ PQfinish(conn);
+ exit(1);
+ }
+
+ /* Remember database OID. */
+ dbinfo[i].oid = strtoul(PQgetvalue(res, 0, 0), NULL, 10);
+
+ PQclear(res);
+
+ /*
+ * Build the replication slot name. The name must not exceed
+ * NAMEDATALEN - 1. This current schema uses a maximum of 36
+ * characters (14 + 10 + 1 + 10 + '\0'). System identifier is included
+ * to reduce the probability of collision. By default, subscription
+ * name is used as replication slot name.
+ */
+ snprintf(replslotname, sizeof(replslotname),
+ "pg_subscriber_%u_%d",
+ dbinfo[i].oid,
+ (int) getpid());
+ dbinfo[i].subname = pg_strdup(replslotname);
+
+ /* Create replication slot on publisher. */
+ if (create_logical_replication_slot(conn, &dbinfo[i], replslotname) != NULL)
+ pg_log_info("create replication slot \"%s\" on publisher", replslotname);
+ else
+ exit(1);
+
+ disconnect_database(conn);
+ }
+
+ /*
+ * Create a temporary logical replication slot to get a consistent LSN.
+ *
+ * This consistent LSN will be used later to advanced the recently created
+ * replication slots. We could probably use the last created replication
+ * slot, however, if this tool decides to support cloning the publisher
+ * (via pg_basebackup -- after creating the replication slots), the
+ * consistent point should be after the pg_basebackup finishes.
+ */
+ conn = connect_database(dbinfo[0].pubconninfo, false);
+ if (conn == NULL)
+ exit(1);
+ snprintf(temp_replslot, sizeof(temp_replslot), "pg_subscriber_%d_tmp",
+ (int) getpid());
+ consistent_lsn = create_logical_replication_slot(conn, &dbinfo[0],
+ temp_replslot);
+
+ /*
+ * Write recovery parameters.
+ *
+ * Despite of the recovery parameters will be written to the subscriber,
+ * use a publisher connection for the follwing recovery functions. The
+ * connection is only used to check the current server version (physical
+ * replica, same server version). The subscriber is not running yet.
+ */
+ recoveryconfcontents = GenerateRecoveryConfig(conn, NULL);
+ appendPQExpBuffer(recoveryconfcontents, "recovery_target_lsn = '%s'\n",
+ consistent_lsn);
+ appendPQExpBuffer(recoveryconfcontents, "recovery_target_inclusive = true\n");
+ appendPQExpBuffer(recoveryconfcontents, "recovery_target_action = promote\n");
+
+ WriteRecoveryConfig(conn, subscriber_dir, recoveryconfcontents);
+ disconnect_database(conn);
+
+ /*
+ * Start subscriber and wait until accepting connections.
+ */
+ if (verbose)
+ pg_log_info("starting the subscriber");
+
+ pg_ctl_cmd = psprintf("\"%s\" start -D \"%s\" -s", pg_ctl_path, subscriber_dir);
+ rc = system(pg_ctl_cmd);
+ pg_ctl_status(pg_ctl_cmd, rc, 1);
+ wait_postmaster_connection(dbinfo[0].subconninfo);
+
+ /*
+ * Waiting the subscriber to be promoted.
+ */
+ wait_for_end_recovery(dbinfo[0].subconninfo);
+
+ /*
+ * Create a publication for each database. This step should be executed
+ * after promoting the subscriber to avoid replicating unnecessary
+ * objects.
+ */
+ for (i = 0; i < num_dbs; i++)
+ {
+ char pubname[NAMEDATALEN];
+
+ /* Connect to publisher. */
+ conn = connect_database(dbinfo[i].pubconninfo, true);
+ if (conn == NULL)
+ exit(1);
+
+ /*
+ * Build the publication name. The name must not exceed NAMEDATALEN -
+ * 1. This current schema uses a maximum of 35 characters (14 + 10 +
+ * '\0').
+ */
+ snprintf(pubname, sizeof(pubname), "pg_subscriber_%u", dbinfo[i].oid);
+ dbinfo[i].pubname = pg_strdup(pubname);
+
+ create_publication(conn, &dbinfo[i]);
+
+ disconnect_database(conn);
+ }
+
+ /*
+ * Create a subscription for each database.
+ */
+ for (i = 0; i < num_dbs; i++)
+ {
+ /* Connect to subscriber. */
+ conn = connect_database(dbinfo[i].subconninfo, true);
+ if (conn == NULL)
+ exit(1);
+
+ create_subscription(conn, &dbinfo[i]);
+
+ /* Set the replication progress to the correct LSN. */
+ set_replication_progress(conn, &dbinfo[i], consistent_lsn);
+
+ /* Enable subscription. */
+ enable_subscription(conn, &dbinfo[i]);
+
+ disconnect_database(conn);
+ }
+
+ /*
+ * The temporary replication slot is no longer required. Drop it.
+ * XXX we might not fail here. Instead, provide a warning so the user
+ * XXX eventually drops the replication slot later.
+ */
+ conn = connect_database(dbinfo[0].pubconninfo, true);
+ if (conn == NULL)
+ exit(1);
+ drop_replication_slot(conn, &dbinfo[0], temp_replslot);
+ disconnect_database(conn);
+
+ /*
+ * Stop the subscriber.
+ */
+ if (verbose)
+ pg_log_info("stopping the subscriber");
+
+ pg_ctl_cmd = psprintf("\"%s\" stop -D \"%s\" -s", pg_ctl_path, subscriber_dir);
+ rc = system(pg_ctl_cmd);
+ pg_ctl_status(pg_ctl_cmd, rc, 0);
+
+ success = true;
+
+ return 0;
+}
diff --git a/src/bin/pg_subscriber/t/001_basic.pl b/src/bin/pg_subscriber/t/001_basic.pl
new file mode 100644
index 0000000000..824e7d7906
--- /dev/null
+++ b/src/bin/pg_subscriber/t/001_basic.pl
@@ -0,0 +1,41 @@
+# Copyright (c) 2022, PostgreSQL Global Development Group
+
+use strict;
+use warnings;
+use Cwd;
+use Config;
+use File::Basename qw(basename dirname);
+use File::Path qw(rmtree);
+use Fcntl qw(:seek);
+use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::Utils;
+use Test::More;
+
+program_help_ok('pg_subscriber');
+program_version_ok('pg_subscriber');
+program_options_handling_ok('pg_subscriber');
+
+my $tempdir = PostgreSQL::Test::Utils::tempdir;
+
+my $node = PostgreSQL::Test::Cluster->new('publisher');
+$node->init(allows_streaming => 'logical');
+$node->start;
+
+$node->command_fails_like(
+ ['pg_subscriber'],
+ qr/no subscriber data directory specified/,
+ 'target directory must be specified');
+$node->command_fails_like(
+ ['pg_subscriber', '-D', $tempdir],
+ qr/no publisher connection string specified/,
+ 'publisher connection string must be specified');
+$node->command_fails_like(
+ ['pg_subscriber', '-D', $tempdir, '-P', 'dbname=postgres'],
+ qr/no subscriber connection string specified/,
+ 'subscriber connection string must be specified');
+$node->command_fails_like(
+ ['pg_subscriber', '-D', $tempdir, '-P', 'dbname=postgres', '-S', 'dbname=postgres'],
+ qr/is not a database cluster directory/,
+ 'directory must be a real database cluster directory');
+
+done_testing();
diff --git a/src/tools/msvc/Mkvcbuild.pm b/src/tools/msvc/Mkvcbuild.pm
index 105f5c72a2..330d312ee6 100644
--- a/src/tools/msvc/Mkvcbuild.pm
+++ b/src/tools/msvc/Mkvcbuild.pm
@@ -55,7 +55,7 @@ my @contrib_excludes = (
# Set of variables for frontend modules
my $frontend_defines = { 'initdb' => 'FRONTEND' };
my @frontend_uselibpq =
- ('pg_amcheck', 'pg_ctl', 'pg_upgrade', 'pgbench', 'psql', 'initdb');
+ ('pg_amcheck', 'pg_ctl', 'pg_upgrade', 'pgbench', 'psql', 'initdb', 'pg_subscriber');
my @frontend_uselibpgport = (
'pg_amcheck', 'pg_archivecleanup',
'pg_test_fsync', 'pg_test_timing',
--
2.30.2
^ permalink raw reply [nested|flat] 5+ messages in thread
* Re: speed up a logical replica setup
@ 2022-03-22 01:25 Andres Freund <[email protected]>
parent: Euler Taveira <[email protected]>
0 siblings, 1 reply; 5+ messages in thread
From: Andres Freund @ 2022-03-22 01:25 UTC (permalink / raw)
To: Euler Taveira <[email protected]>; +Cc: [email protected]
Hi,
On 2022-02-21 09:09:12 -0300, Euler Taveira wrote:
> A new tool called pg_subscriber does this conversion and is tightly integrated
> with Postgres.
Given that this has been submitted just before the last CF and is a patch of
nontrivial size, has't made significant progress ISTM it should be moved to
the next CF?
It currently fails in cfbot, but that's likely just due to Peter's incremental
patch. Perhaps you could make sure the patch still applies and repost?
Greetings,
Andres Freund
^ permalink raw reply [nested|flat] 5+ messages in thread
* Re: speed up a logical replica setup
@ 2022-03-29 12:34 Peter Eisentraut <[email protected]>
parent: Andres Freund <[email protected]>
0 siblings, 1 reply; 5+ messages in thread
From: Peter Eisentraut @ 2022-03-29 12:34 UTC (permalink / raw)
To: Andres Freund <[email protected]>; Euler Taveira <[email protected]>; +Cc: [email protected]
On 22.03.22 02:25, Andres Freund wrote:
> On 2022-02-21 09:09:12 -0300, Euler Taveira wrote:
>> A new tool called pg_subscriber does this conversion and is tightly integrated
>> with Postgres.
>
> Given that this has been submitted just before the last CF and is a patch of
> nontrivial size, has't made significant progress ISTM it should be moved to
> the next CF?
done
^ permalink raw reply [nested|flat] 5+ messages in thread
* Re: speed up a logical replica setup
@ 2022-08-02 18:56 Jacob Champion <[email protected]>
parent: Peter Eisentraut <[email protected]>
0 siblings, 0 replies; 5+ messages in thread
From: Jacob Champion @ 2022-08-02 18:56 UTC (permalink / raw)
To: Peter Eisentraut <[email protected]>; Andres Freund <[email protected]>; Euler Taveira <[email protected]>; +Cc: [email protected]
This entry has been waiting on author input for a while (our current
threshold is roughly two weeks), so I've marked it Returned with
Feedback.
Once you think the patchset is ready for review again, you (or any
interested party) can resurrect the patch entry by visiting
https://commitfest.postgresql.org/38/3556/
and changing the status to "Needs Review", and then changing the
status again to "Move to next CF". (Don't forget the second step;
hopefully we will have streamlined this in the near future!)
Thanks,
--Jacob
^ permalink raw reply [nested|flat] 5+ messages in thread
end of thread, other threads:[~2022-08-02 18:56 UTC | newest]
Thread overview: 5+ messages (download: mbox mbox.gz follow: Atom feed)
-- links below jump to the message on this page --
2020-08-03 23:21 [PATCH v1 2/2] ALTER TABLE ... DETACH CONCURRENTLY Alvaro Herrera <[email protected]>
2022-02-21 12:09 speed up a logical replica setup Euler Taveira <[email protected]>
2022-03-22 01:25 ` Re: speed up a logical replica setup Andres Freund <[email protected]>
2022-03-29 12:34 ` Re: speed up a logical replica setup Peter Eisentraut <[email protected]>
2022-08-02 18:56 ` Re: speed up a logical replica setup Jacob Champion <[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