use_vfd_for_logrep.patch
application/octet-stream
Filename: use_vfd_for_logrep.patch
Type: application/octet-stream
Part: 0
Patch
Same data as JSON:
GET /api/v1/attachments/:id/patch
the parsed metadata as JSON — format, series position, per-file stats; never the diff bytes.
API reference →
Format: unified
| File | + | − |
|---|---|---|
| src/backend/replication/logical/reorderbuffer.c | 36 | 18 |
diff --git a/src/backend/replication/logical/reorderbuffer.c b/src/backend/replication/logical/reorderbuffer.c
index 8ce28ad..c59602c 100644
--- a/src/backend/replication/logical/reorderbuffer.c
+++ b/src/backend/replication/logical/reorderbuffer.c
@@ -103,13 +103,20 @@ typedef struct ReorderBufferTupleCidEnt
CommandId combocid; /* just for debugging */
} ReorderBufferTupleCidEnt;
+/* Virtual file descriptor with file offset tracking */
+typedef struct TXNEntryFile
+{
+ File vfd;
+ off_t curOffset; /* offset for next write or read */
+} TXNEntryFile;
+
/* k-way in-order change iteration support structures */
typedef struct ReorderBufferIterTXNEntry
{
XLogRecPtr lsn;
ReorderBufferChange *change;
ReorderBufferTXN *txn;
- int fd;
+ TXNEntryFile file;
XLogSegNo segno;
} ReorderBufferIterTXNEntry;
@@ -194,7 +201,7 @@ static void ReorderBufferSerializeTXN(ReorderBuffer *rb, ReorderBufferTXN *txn);
static void ReorderBufferSerializeChange(ReorderBuffer *rb, ReorderBufferTXN *txn,
int fd, ReorderBufferChange *change);
static Size ReorderBufferRestoreChanges(ReorderBuffer *rb, ReorderBufferTXN *txn,
- int *fd, XLogSegNo *segno);
+ TXNEntryFile *file, XLogSegNo *segno);
static void ReorderBufferRestoreChange(ReorderBuffer *rb, ReorderBufferTXN *txn,
char *change);
static void ReorderBufferRestoreCleanup(ReorderBuffer *rb, ReorderBufferTXN *txn);
@@ -988,7 +995,8 @@ ReorderBufferIterTXNInit(ReorderBuffer *rb, ReorderBufferTXN *txn)
for (off = 0; off < state->nr_txns; off++)
{
- state->entries[off].fd = -1;
+ state->entries[off].file.vfd = -1;
+ state->entries[off].file.curOffset = 0;
state->entries[off].segno = 0;
}
@@ -1013,7 +1021,7 @@ ReorderBufferIterTXNInit(ReorderBuffer *rb, ReorderBufferTXN *txn)
{
/* serialize remaining changes */
ReorderBufferSerializeTXN(rb, txn);
- ReorderBufferRestoreChanges(rb, txn, &state->entries[off].fd,
+ ReorderBufferRestoreChanges(rb, txn, &state->entries[off].file,
&state->entries[off].segno);
}
@@ -1043,7 +1051,7 @@ ReorderBufferIterTXNInit(ReorderBuffer *rb, ReorderBufferTXN *txn)
/* serialize remaining changes */
ReorderBufferSerializeTXN(rb, cur_txn);
ReorderBufferRestoreChanges(rb, cur_txn,
- &state->entries[off].fd,
+ &state->entries[off].file,
&state->entries[off].segno);
}
cur_change = dlist_head_element(ReorderBufferChange, node,
@@ -1124,7 +1132,7 @@ ReorderBufferIterTXNNext(ReorderBuffer *rb, ReorderBufferIterTXNState *state)
dlist_delete(&change->node);
dlist_push_tail(&state->old_change, &change->node);
- if (ReorderBufferRestoreChanges(rb, entry->txn, &entry->fd,
+ if (ReorderBufferRestoreChanges(rb, entry->txn, &entry->file,
&state->entries[off].segno))
{
/* successfully restored changes from disk */
@@ -1163,8 +1171,8 @@ ReorderBufferIterTXNFinish(ReorderBuffer *rb,
for (off = 0; off < state->nr_txns; off++)
{
- if (state->entries[off].fd != -1)
- CloseTransientFile(state->entries[off].fd);
+ if (state->entries[off].file.vfd != -1)
+ FileClose(state->entries[off].file.vfd);
}
/* free memory we might have "leaked" in the last *Next call */
@@ -2517,11 +2525,12 @@ ReorderBufferSerializeChange(ReorderBuffer *rb, ReorderBufferTXN *txn,
*/
static Size
ReorderBufferRestoreChanges(ReorderBuffer *rb, ReorderBufferTXN *txn,
- int *fd, XLogSegNo *segno)
+ TXNEntryFile *file, XLogSegNo *segno)
{
Size restored = 0;
XLogSegNo last_segno;
dlist_mutable_iter cleanup_iter;
+ File *fd = &file->vfd;
Assert(txn->first_lsn != InvalidXLogRecPtr);
Assert(txn->final_lsn != InvalidXLogRecPtr);
@@ -2562,7 +2571,11 @@ ReorderBufferRestoreChanges(ReorderBuffer *rb, ReorderBufferTXN *txn,
ReorderBufferSerializedPath(path, MyReplicationSlot, txn->xid,
*segno);
- *fd = OpenTransientFile(path, O_RDONLY | PG_BINARY);
+ *fd = PathNameOpenFile(path, O_RDONLY | PG_BINARY);
+
+ /* Better reset the offset in case of failure as well */
+ file->curOffset = 0;
+
if (*fd < 0 && errno == ENOENT)
{
*fd = -1;
@@ -2582,14 +2595,14 @@ ReorderBufferRestoreChanges(ReorderBuffer *rb, ReorderBufferTXN *txn,
* end of this file.
*/
ReorderBufferSerializeReserve(rb, sizeof(ReorderBufferDiskChange));
- pgstat_report_wait_start(WAIT_EVENT_REORDER_BUFFER_READ);
- readBytes = read(*fd, rb->outbuf, sizeof(ReorderBufferDiskChange));
- pgstat_report_wait_end();
+ readBytes = FileRead(file->vfd, rb->outbuf,
+ sizeof(ReorderBufferDiskChange),
+ file->curOffset, WAIT_EVENT_REORDER_BUFFER_READ);
/* eof */
if (readBytes == 0)
{
- CloseTransientFile(*fd);
+ FileClose(*fd);
*fd = -1;
(*segno)++;
continue;
@@ -2605,16 +2618,19 @@ ReorderBufferRestoreChanges(ReorderBuffer *rb, ReorderBufferTXN *txn,
readBytes,
(uint32) sizeof(ReorderBufferDiskChange))));
+ file->curOffset += readBytes;
+
ondisk = (ReorderBufferDiskChange *) rb->outbuf;
ReorderBufferSerializeReserve(rb,
sizeof(ReorderBufferDiskChange) + ondisk->size);
ondisk = (ReorderBufferDiskChange *) rb->outbuf;
- pgstat_report_wait_start(WAIT_EVENT_REORDER_BUFFER_READ);
- readBytes = read(*fd, rb->outbuf + sizeof(ReorderBufferDiskChange),
- ondisk->size - sizeof(ReorderBufferDiskChange));
- pgstat_report_wait_end();
+ readBytes = FileRead(file->vfd,
+ rb->outbuf + sizeof(ReorderBufferDiskChange),
+ ondisk->size - sizeof(ReorderBufferDiskChange),
+ file->curOffset,
+ WAIT_EVENT_REORDER_BUFFER_READ);
if (readBytes < 0)
ereport(ERROR,
@@ -2627,6 +2643,8 @@ ReorderBufferRestoreChanges(ReorderBuffer *rb, ReorderBufferTXN *txn,
readBytes,
(uint32) (ondisk->size - sizeof(ReorderBufferDiskChange)))));
+ file->curOffset += readBytes;
+
/*
* ok, read a full change from disk, now restore it into proper
* in-memory format