public inbox for [email protected]help / color / mirror / Atom feed
[PATCH] Fix timeline-tracking failure while sending a historic timeline 2+ messages / 2 participants [nested] [flat]
* [PATCH] Fix timeline-tracking failure while sending a historic timeline @ 2021-01-05 04:34 Kyotaro Horiguchi <[email protected]> 0 siblings, 0 replies; 2+ messages in thread From: Kyotaro Horiguchi @ 2021-01-05 04:34 UTC (permalink / raw) Walsender should track timeline switches while sending a historic timeline. Regain that behavior, which was broken in PG13, by a thinko of 709d003fbd. Backpatch to PG13. --- src/backend/replication/walsender.c | 2 +- src/test/perl/PostgresNode.pm | 33 ++++++++++++++++++ src/test/perl/TestLib.pm | 16 ++++++--- src/test/recovery/t/001_stream_rep.pl | 41 ++++++++++++++++++++++- src/test/recovery/t/019_replslot_limit.pl | 37 ++++---------------- 5 files changed, 92 insertions(+), 37 deletions(-) diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c index 7f87eb7f19..04f6c3ebb4 100644 --- a/src/backend/replication/walsender.c +++ b/src/backend/replication/walsender.c @@ -2478,7 +2478,7 @@ WalSndSegmentOpen(XLogReaderState *state, XLogSegNo nextSegNo, XLogSegNo endSegNo; XLByteToSeg(sendTimeLineValidUpto, endSegNo, state->segcxt.ws_segsize); - if (state->seg.ws_segno == endSegNo) + if (nextSegNo == endSegNo) *tli_p = sendTimeLineNextTLI; } diff --git a/src/test/perl/PostgresNode.pm b/src/test/perl/PostgresNode.pm index 980f1f1533..687aa3ac88 100644 --- a/src/test/perl/PostgresNode.pm +++ b/src/test/perl/PostgresNode.pm @@ -2138,6 +2138,39 @@ sub pg_recvlogical_upto =pod +=item $node->current_log_position() + +Return the current position of server log. + +=cut + +sub current_log_position +{ + my $self = shift; + + return (stat $self->logfile)[7]; +} + +=pod + +=item $node->find_in_log($pattern, $startpos) + +Returns whether the $pattern occurs after $startpos in the server log. + +=cut + +sub find_in_log +{ + my ($self, $pattern, $startpos) = @_; + + $startpos = 0 unless defined $startpos; + my $log = TestLib::slurp_file($self->logfile, $startpos); + + return $log =~ m/$pattern/; +} + +=pod + =back =cut diff --git a/src/test/perl/TestLib.pm b/src/test/perl/TestLib.pm index a7490d2ce7..a0ce9521e2 100644 --- a/src/test/perl/TestLib.pm +++ b/src/test/perl/TestLib.pm @@ -402,30 +402,38 @@ sub slurp_dir =pod -=item slurp_file(filename) +=item slurp_file(filename, pos) -Return the full contents of the specified file. +Return the contents after pos of the specified file. +Reutrns the full contents if pos is omitted. =cut sub slurp_file { - my ($filename) = @_; + my ($filename, $from) = @_; local $/; my $contents; + + $from = 0 unless defined $from; + if ($Config{osname} ne 'MSWin32') { open(my $in, '<', $filename) or die "could not read \"$filename\": $!"; + seek($in, $from, 0) + or die "could not seek \"$filename\" to $from: $!"; $contents = <$in>; close $in; } else { my $fHandle = createFile($filename, "r", "rwd") - or die "could not open \"$filename\": $^E"; + or die "could not open \"$filename\": $^E\n"; OsFHandleOpen(my $fh = IO::Handle->new(), $fHandle, 'r') or die "could not read \"$filename\": $^E\n"; + seek($fh, $from, 0) + or die "could not seek \"$filename\" to $from: $^E\n"; $contents = <$fh>; CloseHandle($fHandle) or die "could not close \"$filename\": $^E\n"; diff --git a/src/test/recovery/t/001_stream_rep.pl b/src/test/recovery/t/001_stream_rep.pl index 778f11b28b..8d2b24fe55 100644 --- a/src/test/recovery/t/001_stream_rep.pl +++ b/src/test/recovery/t/001_stream_rep.pl @@ -2,8 +2,9 @@ use strict; use warnings; use PostgresNode; +use Time::HiRes qw(usleep); use TestLib; -use Test::More tests => 36; +use Test::More tests => 37; # Initialize master node my $node_master = get_new_node('master'); @@ -409,3 +410,41 @@ ok( ($phys_restart_lsn_pre cmp $phys_restart_lsn_post) == 0, my $master_data = $node_master->data_dir; ok(!-f "$master_data/pg_wal/$segment_removed", "WAL segment $segment_removed recycled after physical slot advancing"); + +# +# Check if timeline-increment works while reading a historic timeline. +my $node_primary_2 = get_new_node('primary_2'); +# archiving is needed to create .paritial segment +$node_primary_2->init(allows_streaming => 1, has_archiving => 1); +$node_primary_2->start; +$node_primary_2->backup($backup_name); +my $node_standby_3 = get_new_node('standby_3'); +$node_standby_3->init_from_backup($node_primary_2, $backup_name, + has_streaming => 1); +$node_primary_2->stop; +$node_primary_2->set_standby_mode; # increment primary timeline +$node_primary_2->start; +$node_primary_2->promote; +my $logstart = $node_standby_3->current_log_position(); +$node_standby_3->start; + +my $success = 0; +for (my $i = 0 ; $i < 1000; $i++) +{ + if ($node_standby_3->find_in_log( + "requested WAL segment [0-9A-F]+ has already been removed", + $logstart)) + { + last; + } + elsif ($node_standby_3->find_in_log( + "End of WAL reached on timeline", + $logstart)) + { + $success = 1; + last; + } + usleep(100_000); +} + +ok($success, 'Timeline increment while reading a historic timeline'); diff --git a/src/test/recovery/t/019_replslot_limit.pl b/src/test/recovery/t/019_replslot_limit.pl index a7231dcd47..8b3c5de057 100644 --- a/src/test/recovery/t/019_replslot_limit.pl +++ b/src/test/recovery/t/019_replslot_limit.pl @@ -165,19 +165,17 @@ $node_master->wait_for_catchup($node_standby, 'replay', $start_lsn); $node_standby->stop; -ok( !find_in_log( - $node_standby, - "requested WAL segment [0-9A-F]+ has already been removed"), +ok( !$node_standby->find_in_log( + "requested WAL segment [0-9A-F]+ has already been removed"), 'check that required WAL segments are still available'); # Advance WAL again, the slot loses the oldest segment. -my $logstart = get_log_size($node_master); +my $logstart = $node_master->current_log_position(); advance_wal($node_master, 7); $node_master->safe_psql('postgres', "CHECKPOINT;"); # WARNING should be issued -ok( find_in_log( - $node_master, +ok( $node_master->find_in_log( "invalidating slot \"rep1\" because its restart_lsn [0-9A-F/]+ exceeds max_slot_wal_keep_size", $logstart), 'check that the warning is logged'); @@ -190,14 +188,13 @@ is($result, "rep1|f|t|lost|", 'check that the slot became inactive and the state "lost" persists'); # The standby no longer can connect to the master -$logstart = get_log_size($node_standby); +$logstart = $node_standby->current_log_position(); $node_standby->start; my $failed = 0; for (my $i = 0; $i < 10000; $i++) { - if (find_in_log( - $node_standby, + if ($node_standby->find_in_log( "requested WAL segment [0-9A-F]+ has already been removed", $logstart)) { @@ -264,25 +261,3 @@ sub advance_wal } return; } - -# return the size of logfile of $node in bytes -sub get_log_size -{ - my ($node) = @_; - - return (stat $node->logfile)[7]; -} - -# find $pat in logfile of $node after $off-th byte -sub find_in_log -{ - my ($node, $pat, $off) = @_; - - $off = 0 unless defined $off; - my $log = TestLib::slurp_file($node->logfile); - return 0 if (length($log) <= $off); - - $log = substr($log, $off); - - return $log =~ m/$pat/; -} -- 2.27.0 ----Next_Part(Wed_Jan__6_10_48_04_2021_104)---- ^ permalink raw reply [nested|flat] 2+ messages in thread
* [PATCH va1 2/2] Mega-WIP: Optimized out/send path for printtup @ 2026-05-06 15:44 Andres Freund <[email protected]> 0 siblings, 0 replies; 2+ messages in thread From: Andres Freund @ 2026-05-06 15:44 UTC (permalink / raw) Discussion: https://postgr.es/m/[email protected] --- src/include/nodes/miscnodes.h | 12 +++++ src/backend/access/common/printtup.c | 35 +++++++++++-- src/backend/utils/adt/int.c | 73 +++++++++++++++++++++++++--- src/backend/utils/adt/varlena.c | 40 ++++++++++++++- 4 files changed, 148 insertions(+), 12 deletions(-) diff --git a/src/include/nodes/miscnodes.h b/src/include/nodes/miscnodes.h index ec833001ab0..b3c189a5c0c 100644 --- a/src/include/nodes/miscnodes.h +++ b/src/include/nodes/miscnodes.h @@ -54,4 +54,16 @@ typedef struct ErrorSaveContext ((escontext) != NULL && IsA(escontext, ErrorSaveContext) && \ ((ErrorSaveContext *) (escontext))->error_occurred) + +/* + * Type optionally passed to input/receive/output/send functions that allows + * those functions to opt into more efficient ways of performing their work + * (mainly reducing allocations & copies). + */ +typedef struct InOutContext +{ + NodeTag type; + StringInfo buf; +} InOutContext; + #endif /* MISCNODES_H */ diff --git a/src/backend/access/common/printtup.c b/src/backend/access/common/printtup.c index 6fa93a6798a..2e3eb8f56d3 100644 --- a/src/backend/access/common/printtup.c +++ b/src/backend/access/common/printtup.c @@ -63,6 +63,7 @@ typedef struct int nattrs; PrinttupAttrInfo *myinfo; /* Cached info about each attr */ StringInfoData buf; /* output buffer (*not* in tmpcontext) */ + InOutContext inout; /* FunctionCallInfo->context data */ MemoryContext tmpcontext; /* Memory context for per-row workspace */ } DR_printtup; @@ -142,6 +143,9 @@ printtup_startup(DestReceiver *self, int operation, TupleDesc typeinfo) FetchPortalTargetList(portal), portal->formats); + myState->inout.type = T_InOutContext; + myState->inout.buf = &myState->buf; + /* ---------------- * We could set up the derived attr info at this time, but we postpone it * until the first call of printtup, for 2 reasons: @@ -297,6 +301,15 @@ printtup_prepare_info(DR_printtup *myState, TupleDesc typeinfo, int numAttrs) /* both out and send funcs have one argument */ thisState->outstate = palloc0(SizeForFunctionCallInfo(1)); thisState->outstate->flinfo = &thisState->finfo; + + /* + * The idea here is that output functions can optionally use more + * efficient paths if they see that the context is InOutContext, by + * directly appending correctly formatted output into the output + * buffer. + */ + thisState->outstate->context = (Node *) &myState->inout; + thisState->outstate->nargs = 1; } } @@ -369,7 +382,13 @@ printtup(TupleTableSlot *slot, DestReceiver *self) outputstr = DatumGetCString(FunctionCallInvoke(thisState->outstate)); Assert(!thisState->outstate->isnull); - pq_sendcountedtext(buf, outputstr, strlen(outputstr)); + + /* + * If outputstr == NULL, the output function directly appended a + * correctly formatted message. + */ + if (outputstr) + pq_sendcountedtext(buf, outputstr, strlen(outputstr)); } else { @@ -378,9 +397,17 @@ printtup(TupleTableSlot *slot, DestReceiver *self) outputbytes = DatumGetByteaP(FunctionCallInvoke(thisState->outstate)); Assert(!thisState->outstate->isnull); - pq_sendint32(buf, VARSIZE(outputbytes) - VARHDRSZ); - pq_sendbytes(buf, VARDATA(outputbytes), - VARSIZE(outputbytes) - VARHDRSZ); + + /* + * If outputbytes == NULL, the send function directly appended a + * correctly formatted message. + */ + if (outputbytes) + { + pq_sendint32(buf, VARSIZE(outputbytes) - VARHDRSZ); + pq_sendbytes(buf, VARDATA(outputbytes), + VARSIZE(outputbytes) - VARHDRSZ); + } } } diff --git a/src/backend/utils/adt/int.c b/src/backend/utils/adt/int.c index 4c894a49d5d..1b7b5b5246c 100644 --- a/src/backend/utils/adt/int.c +++ b/src/backend/utils/adt/int.c @@ -327,10 +327,54 @@ Datum int4out(PG_FUNCTION_ARGS) { int32 arg1 = PG_GETARG_INT32(0); - char *result = (char *) palloc(12); /* sign, 10 digits, '\0' */ + int maxlen = 12; /* sign, 10 digits, '\0' */ - pg_ltoa(arg1, result); - PG_RETURN_CSTRING(result); + if (fcinfo->context && IsA(fcinfo->context, InOutContext)) + { + /* + * Optimized path for output functions called as part of a larger + * ouput. + * + * FIXME: A good chunk of this should obviously be in helper + * functions. + */ + InOutContext *inout = castNode(InOutContext, fcinfo->context); + StringInfo buf = inout->buf; + int prev_buflen; + int len; + uint32 len_net; + + /* reserve space for length and the max string length */ + enlargeStringInfo(buf, sizeof(uint32) + maxlen); + + /* reserve space for length, to be filled out later */ + prev_buflen = buf->len; + buf->len += sizeof(uint32); + + /* + * Construct string directly in buffer, we don't have to care about + * encoding conversions, because we assume that every encoding + * embodies ascii (XXX: Is that actually true with client encodings?). + */ + len = pg_ltoa(arg1, buf->data + buf->len); + buf->len += len; + + /* update the previously reserved length */ + len_net = pg_hton32(len); + memcpy(&buf->data[prev_buflen], &len_net, sizeof(uint32)); + + PG_RETURN_VOID(); + } + else + { + /* + * Fallback path called in any other context. + */ + char *result = (char *) palloc(maxlen); + + pg_ltoa(arg1, result); + PG_RETURN_CSTRING(result); + } } /* @@ -351,11 +395,26 @@ Datum int4send(PG_FUNCTION_ARGS) { int32 arg1 = PG_GETARG_INT32(0); - StringInfoData buf; - pq_begintypsend(&buf); - pq_sendint32(&buf, arg1); - PG_RETURN_BYTEA_P(pq_endtypsend(&buf)); + if (fcinfo->context && IsA(fcinfo->context, InOutContext)) + { + InOutContext *inout = castNode(InOutContext, fcinfo->context); + + /* length of data */ + pq_sendint32(inout->buf, 4); + /* data itself */ + pq_sendint32(inout->buf, arg1); + + PG_RETURN_VOID(); + } + else + { + StringInfoData buf; + + pq_begintypsend(&buf); + pq_sendint32(&buf, arg1); + PG_RETURN_BYTEA_P(pq_endtypsend(&buf)); + } } diff --git a/src/backend/utils/adt/varlena.c b/src/backend/utils/adt/varlena.c index c0ff51bd2fc..09913bc01f5 100644 --- a/src/backend/utils/adt/varlena.c +++ b/src/backend/utils/adt/varlena.c @@ -290,7 +290,45 @@ textout(PG_FUNCTION_ARGS) { Datum txt = PG_GETARG_DATUM(0); - PG_RETURN_CSTRING(TextDatumGetCString(txt)); + if (fcinfo->context && IsA(fcinfo->context, InOutContext)) + { + StringInfo buf = castNode(InOutContext, fcinfo->context)->buf; + text *tunpacked = pg_detoast_datum_packed(DatumGetPointer(txt)); + int len = VARSIZE_ANY_EXHDR(tunpacked); + char *data = VARDATA_ANY(tunpacked); + char *data_converted; + size_t data_len; + + /* + * Convert text output to the right encoding. For efficiency, this + * should really happen directly into buf. For that we would have to + * reserve space for the length first and fill it out after + * conversion. + * + * FIXME: Obviously we would need helpers for this too. + */ + data_converted = pg_server_to_client(data, len); + + if (data == data_converted) + data_len = len; + else + data_len = strlen(data_converted); + + /* length */ + pq_sendint32(buf, data_len); + + /* actual data */ + appendBinaryStringInfoNT(buf, data_converted, data_len); + + if (tunpacked != DatumGetPointer(txt)) + pfree(tunpacked); + + PG_RETURN_VOID(); + } + else + { + PG_RETURN_CSTRING(TextDatumGetCString(txt)); + } } /* -- 2.46.0.519.g2e7b89e038 --benhmb75h34eqzc4-- ^ permalink raw reply [nested|flat] 2+ messages in thread
end of thread, other threads:[~2026-05-06 15:44 UTC | newest] Thread overview: 2+ messages (download: mbox mbox.gz follow: Atom feed) -- links below jump to the message on this page -- 2021-01-05 04:34 [PATCH] Fix timeline-tracking failure while sending a historic timeline Kyotaro Horiguchi <[email protected]> 2026-05-06 15:44 [PATCH va1 2/2] Mega-WIP: Optimized out/send path for printtup Andres Freund <[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