Skip to content

Navigation Menu

Sign in
Appearance settings

Search code, repositories, users, issues, pull requests...

Provide feedback

We read every piece of feedback, and take your input very seriously.

Saved searches

Use saved searches to filter your results more quickly
Appearance settings

Commit 07082b0

Browse filesBrowse files
alvherrealvherre
committed
Fix bogus completion tag usage in walsender
Since commit fd5942c (2012, 9.3-era), walsender has been sending completion tags for certain replication commands twice -- and they're not even consistent. Apparently neither libpq nor JDBC have a problem with it, but it's not kosher. Fix by remove the EndCommand() call in the common code path for them all, and inserting specific calls to EndReplicationCommand() specifically in those places where it's needed. EndReplicationCommand() is a new simple function to send the completion tag for replication commands. Do this instead of sending a generic SELECT completion tag for them all, which was also pretty bogus (if innocuous). While at it, change StartReplication() to use EndReplicationCommand() instead of pg_puttextmessage(). In commit 2f96613, I failed to realize that replication commands are not close-enough kin of regular SQL commands, so the DROP_REPLICATION_SLOT tag I added is undeserved and a type pun. Take it out. Backpatch to 13, where the latter commit appeared. The duplicate tag has been sent since 9.3, but since nothing is broken, it doesn't seem worth fixing. Per complaints from Tom Lane. Discussion: https://postgr.es/m/1347966.1600195735@sss.pgh.pa.us
1 parent 44fc6e2 commit 07082b0
Copy full SHA for 07082b0

File tree

Expand file treeCollapse file tree

4 files changed

+34
-14
lines changed
Filter options
Expand file treeCollapse file tree

4 files changed

+34
-14
lines changed

‎src/backend/replication/walsender.c

Copy file name to clipboardExpand all lines: src/backend/replication/walsender.c
+21-13Lines changed: 21 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -799,7 +799,7 @@ StartReplication(StartReplicationCmd *cmd)
799799
}
800800

801801
/* Send CommandComplete message */
802-
pq_puttextmessage('C', "START_STREAMING");
802+
EndReplicationCommand("START_STREAMING");
803803
}
804804

805805
/*
@@ -1122,11 +1122,7 @@ CreateReplicationSlot(CreateReplicationSlotCmd *cmd)
11221122
static void
11231123
DropReplicationSlot(DropReplicationSlotCmd *cmd)
11241124
{
1125-
QueryCompletion qc;
1126-
11271125
ReplicationSlotDrop(cmd->slotname, !cmd->wait);
1128-
SetQueryCompletion(&qc, CMDTAG_DROP_REPLICATION_SLOT, 0);
1129-
EndCommand(&qc, DestRemote, false);
11301126
}
11311127

11321128
/*
@@ -1517,9 +1513,9 @@ exec_replication_command(const char *cmd_string)
15171513
{
15181514
int parse_rc;
15191515
Node *cmd_node;
1516+
const char *cmdtag;
15201517
MemoryContext cmd_context;
15211518
MemoryContext old_context;
1522-
QueryCompletion qc;
15231519

15241520
/*
15251521
* If WAL sender has been told that shutdown is getting close, switch its
@@ -1619,51 +1615,67 @@ exec_replication_command(const char *cmd_string)
16191615
switch (cmd_node->type)
16201616
{
16211617
case T_IdentifySystemCmd:
1618+
cmdtag = "IDENTIFY_SYSTEM";
16221619
IdentifySystem();
1620+
EndReplicationCommand(cmdtag);
16231621
break;
16241622

16251623
case T_BaseBackupCmd:
1626-
PreventInTransactionBlock(true, "BASE_BACKUP");
1624+
cmdtag = "BASE_BACKUP";
1625+
PreventInTransactionBlock(true, cmdtag);
16271626
SendBaseBackup((BaseBackupCmd *) cmd_node);
1627+
EndReplicationCommand(cmdtag);
16281628
break;
16291629

16301630
case T_CreateReplicationSlotCmd:
1631+
cmdtag = "CREATE_REPLICATION_SLOT";
16311632
CreateReplicationSlot((CreateReplicationSlotCmd *) cmd_node);
1633+
EndReplicationCommand(cmdtag);
16321634
break;
16331635

16341636
case T_DropReplicationSlotCmd:
1637+
cmdtag = "DROP_REPLICATION_SLOT";
16351638
DropReplicationSlot((DropReplicationSlotCmd *) cmd_node);
1639+
EndReplicationCommand(cmdtag);
16361640
break;
16371641

16381642
case T_StartReplicationCmd:
16391643
{
16401644
StartReplicationCmd *cmd = (StartReplicationCmd *) cmd_node;
16411645

1642-
PreventInTransactionBlock(true, "START_REPLICATION");
1646+
cmdtag = "START_REPLICATION";
1647+
PreventInTransactionBlock(true, cmdtag);
16431648

16441649
if (cmd->kind == REPLICATION_KIND_PHYSICAL)
16451650
StartReplication(cmd);
16461651
else
16471652
StartLogicalReplication(cmd);
16481653

1654+
/* callees already sent their own completion message */
1655+
16491656
Assert(xlogreader != NULL);
16501657
break;
16511658
}
16521659

16531660
case T_TimeLineHistoryCmd:
1654-
PreventInTransactionBlock(true, "TIMELINE_HISTORY");
1661+
cmdtag = "TIMELINE_HISTORY";
1662+
PreventInTransactionBlock(true, cmdtag);
16551663
SendTimeLineHistory((TimeLineHistoryCmd *) cmd_node);
1664+
EndReplicationCommand(cmdtag);
16561665
break;
16571666

16581667
case T_VariableShowStmt:
16591668
{
16601669
DestReceiver *dest = CreateDestReceiver(DestRemoteSimple);
16611670
VariableShowStmt *n = (VariableShowStmt *) cmd_node;
16621671

1672+
cmdtag = "SHOW";
1673+
16631674
/* syscache access needs a transaction environment */
16641675
StartTransactionCommand();
16651676
GetPGVariable(n->name, dest);
16661677
CommitTransactionCommand();
1678+
EndReplicationCommand(cmdtag);
16671679
}
16681680
break;
16691681

@@ -1676,10 +1688,6 @@ exec_replication_command(const char *cmd_string)
16761688
MemoryContextSwitchTo(old_context);
16771689
MemoryContextDelete(cmd_context);
16781690

1679-
/* Send CommandComplete message */
1680-
SetQueryCompletion(&qc, CMDTAG_SELECT, 0);
1681-
EndCommand(&qc, DestRemote, true);
1682-
16831691
/* Report to pgstat that this process is now idle */
16841692
pgstat_report_activity(STATE_IDLE, NULL);
16851693
debug_query_string = NULL;

‎src/backend/tcop/dest.c

Copy file name to clipboardExpand all lines: src/backend/tcop/dest.c
+12Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -211,6 +211,18 @@ EndCommand(const QueryCompletion *qc, CommandDest dest, bool force_undecorated_o
211211
}
212212
}
213213

214+
/* ----------------
215+
* EndReplicationCommand - stripped down version of EndCommand
216+
*
217+
* For use by replication commands.
218+
* ----------------
219+
*/
220+
void
221+
EndReplicationCommand(const char *commandTag)
222+
{
223+
pq_putmessage('C', commandTag, strlen(commandTag) + 1);
224+
}
225+
214226
/* ----------------
215227
* NullCommand - tell dest that an empty query string was recognized
216228
*

‎src/include/tcop/cmdtaglist.h

Copy file name to clipboardExpand all lines: src/include/tcop/cmdtaglist.h
-1Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -157,7 +157,6 @@ PG_CMDTAG(CMDTAG_DROP_OWNED, "DROP OWNED", true, false, false)
157157
PG_CMDTAG(CMDTAG_DROP_POLICY, "DROP POLICY", true, false, false)
158158
PG_CMDTAG(CMDTAG_DROP_PROCEDURE, "DROP PROCEDURE", true, false, false)
159159
PG_CMDTAG(CMDTAG_DROP_PUBLICATION, "DROP PUBLICATION", true, false, false)
160-
PG_CMDTAG(CMDTAG_DROP_REPLICATION_SLOT, "DROP REPLICATION SLOT", false, false, false)
161160
PG_CMDTAG(CMDTAG_DROP_ROLE, "DROP ROLE", false, false, false)
162161
PG_CMDTAG(CMDTAG_DROP_ROUTINE, "DROP ROUTINE", true, false, false)
163162
PG_CMDTAG(CMDTAG_DROP_RULE, "DROP RULE", true, false, false)

‎src/include/tcop/dest.h

Copy file name to clipboardExpand all lines: src/include/tcop/dest.h
+1Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -139,6 +139,7 @@ extern void BeginCommand(CommandTag commandTag, CommandDest dest);
139139
extern DestReceiver *CreateDestReceiver(CommandDest dest);
140140
extern void EndCommand(const QueryCompletion *qc, CommandDest dest,
141141
bool force_undecorated_output);
142+
extern void EndReplicationCommand(const char *commandTag);
142143

143144
/* Additional functions that go with destination management, more or less. */
144145

0 commit comments

Comments
0 (0)
Morty Proxy This is a proxified and sanitized view of the page, visit original site.