rust/neon
1
0
Fork 0
mirror of https://github.com/neondatabase/neon.git synced 2026-05-17 05:00:38 +00:00
Code Issues Packages Projects Releases Wiki Activity

Update the relsize cache rather than forget it at end of index build

Browse Source 
This greatly reduces the cases where we make a request to the
pageserver with a very recent LSN. Those cases are slow because the
pageserver needs to wait for the WAL to arrive. This speeds up the
Postgres pg_regress and isolation tests greatly.
This commit is contained in:
Heikki Linnakangas
2025-07-09 17:21:06 +03:00
parent 60d87966b8
commit ed4652b65b
5 changed files with 54 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
pgxn/neon/communicator/src/neon_request.rs
View File

@@ -30,7 +30,7 @@ pub enum NeonIORequest {
RelUnlink(CRelUnlinkRequest),
// Other requests
ForgetCache(CForgetCacheRequest),
UpdateCachedRelSize(CUpdateCachedRelSizeRequest),
}
#[repr(C)]
@@ -75,7 +75,7 @@ impl NeonIORequest {
RelCreate(req) => req.request_id,
RelTruncate(req) => req.request_id,
RelUnlink(req) => req.request_id,
ForgetCache(req) => req.request_id,
UpdateCachedRelSize(req) => req.request_id,
}
}
}
@@ -382,7 +382,7 @@ impl CRelUnlinkRequest {
#[repr(C)]
#[derive(Copy, Clone, Debug)]
pub struct CForgetCacheRequest {
pub struct CUpdateCachedRelSizeRequest {
pub request_id: u64,
pub spc_oid: COid,
pub db_oid: COid,
@@ -392,7 +392,7 @@ pub struct CForgetCacheRequest {
pub lsn: CLsn,
}
impl CForgetCacheRequest {
impl CUpdateCachedRelSizeRequest {
pub fn reltag(&self) -> page_api::RelTag {
page_api::RelTag {
spcnode: self.spc_oid,

10
pgxn/neon/communicator/src/worker_process/main_loop.rs
View File

@@ -505,10 +505,10 @@ impl<'t> CommunicatorWorkerProcessStruct<'t> {
self.cache.forget_rel(&req.reltag(), None, Lsn(req.lsn));
NeonIOResult::WriteOK
}
NeonIORequest::ForgetCache(req) => {
NeonIORequest::UpdateCachedRelSize(req) => {
// TODO: need to grab an io-in-progress lock for this? I guess not
self.cache
.forget_rel(&req.reltag(), Some(req.nblocks), Lsn(req.lsn));
.remember_rel_size(&req.reltag(), req.nblocks, Lsn(req.lsn));
NeonIOResult::WriteOK
}
}
@@ -597,6 +597,12 @@ impl<'t> CommunicatorWorkerProcessStruct<'t> {
);
return Err(-1);
}
info!(
"received getpage response for blocks {:?} in rel {:?} lsns {}",
block_numbers, rel, read_lsn
);
for (page_image, (blkno, _lsn, dest, _guard)) in
resp.page_images.into_iter().zip(cache_misses)
{

25
pgxn/neon/communicator_new.c
View File

@@ -1107,6 +1107,9 @@ communicator_new_rel_create(NRelFileInfo rinfo, ForkNumber forkNum, XLogRecPtr l
};
NeonIOResult result;
/* FIXME: see `request_lsns` in main_loop.rs for why this is needed */
XLogSetAsyncXactLSN(lsn);
perform_request(&request, &result);
switch (result.tag)
{
@@ -1141,6 +1144,9 @@ communicator_new_rel_truncate(NRelFileInfo rinfo, ForkNumber forkNum, BlockNumbe
};
NeonIOResult result;
/* FIXME: see `request_lsns` in main_loop.rs for why this is needed */
XLogSetAsyncXactLSN(lsn);
perform_request(&request, &result);
switch (result.tag)
{
@@ -1174,6 +1180,9 @@ communicator_new_rel_unlink(NRelFileInfo rinfo, ForkNumber forkNum, XLogRecPtr l
};
NeonIOResult result;
/* FIXME: see `request_lsns` in main_loop.rs for why this is needed */
XLogSetAsyncXactLSN(lsn);
perform_request(&request, &result);
switch (result.tag)
{
@@ -1192,11 +1201,11 @@ communicator_new_rel_unlink(NRelFileInfo rinfo, ForkNumber forkNum, XLogRecPtr l
}
void
communicator_new_forget_cache(NRelFileInfo rinfo, ForkNumber forkNum, BlockNumber nblocks, XLogRecPtr lsn)
communicator_new_update_cached_rel_size(NRelFileInfo rinfo, ForkNumber forkNum, BlockNumber nblocks, XLogRecPtr lsn)
{
NeonIORequest request = {
.tag = NeonIORequest_ForgetCache,
.forget_cache = {
.tag = NeonIORequest_UpdateCachedRelSize,
.update_cached_rel_size = {
.request_id = assign_request_id(),
.spc_oid = NInfoGetSpcOid(rinfo),
.db_oid = NInfoGetDbOid(rinfo),
@@ -1216,11 +1225,11 @@ communicator_new_forget_cache(NRelFileInfo rinfo, ForkNumber forkNum, BlockNumbe
case NeonIOResult_Error:
ereport(ERROR,
(errcode_for_file_access(),
errmsg("could not forget cache for rel %u/%u/%u.%u: %s",
errmsg("could not update cached size for rel %u/%u/%u.%u: %s",
RelFileInfoFmt(rinfo), forkNum, pg_strerror(result.error))));
break;
default:
elog(ERROR, "unexpected result for ForgetCache operation: %d", result.tag);
elog(ERROR, "unexpected result for UpdateCachedRelSize operation: %d", result.tag);
break;
}
}
@@ -1338,11 +1347,11 @@ print_neon_io_request(NeonIORequest *request)
r->spc_oid, r->db_oid, r->rel_number, r->fork_number);
return buf;
}
case NeonIORequest_ForgetCache:
case NeonIORequest_UpdateCachedRelSize:
{
CForgetCacheRequest *r = &request->forget_cache;
CUpdateCachedRelSizeRequest *r = &request->update_cached_rel_size;
snprintf(buf, sizeof(buf), "ForgetCache: req " UINT64_FORMAT " rel %u/%u/%u.%u blocks: %u",
snprintf(buf, sizeof(buf), "UpdateCachedRelSize: req " UINT64_FORMAT " rel %u/%u/%u.%u blocks: %u",
r->request_id,
r->spc_oid, r->db_oid, r->rel_number, r->fork_number,
r->nblocks);

2
pgxn/neon/communicator_new.h
View File

@@ -52,7 +52,7 @@ extern void communicator_new_rel_zeroextend(NRelFileInfo rinfo, ForkNumber forkN
extern void communicator_new_rel_create(NRelFileInfo rinfo, ForkNumber forkNum, XLogRecPtr lsn);
extern void communicator_new_rel_truncate(NRelFileInfo rinfo, ForkNumber forkNum, BlockNumber nblocks, XLogRecPtr lsn);
extern void communicator_new_rel_unlink(NRelFileInfo rinfo, ForkNumber forkNum, XLogRecPtr lsn);
extern void communicator_new_forget_cache(NRelFileInfo rinfo, ForkNumber forkNum, BlockNumber nblocks, XLogRecPtr lsn);
extern void communicator_new_update_cached_rel_size(NRelFileInfo rinfo, ForkNumber forkNum, BlockNumber nblocks, XLogRecPtr lsn);
/* other functions */
extern int32 communicator_new_approximate_working_set_size_seconds(time_t duration, bool reset);

26
pgxn/neon/pagestore_smgr.c
View File

@@ -531,6 +531,29 @@ neon_get_write_lsn(void)
else
lsn = GetXLogInsertRecPtr();
/*
* If the insert LSN points to just after page header, round it down to
* the beginning of the page, because the page header might not have been
* inserted to the WAL yet, and if we tried to flush it, the WAL flushing
* code gets upset.
*/
{
int segoff;
segoff = XLogSegmentOffset(lsn, wal_segment_size);
if (segoff == SizeOfXLogLongPHD)
{
lsn = lsn - segoff;
}
else
{
int offset = lsn % XLOG_BLCKSZ;
if (offset == SizeOfXLogShortPHD)
lsn = lsn - offset;
}
}
return lsn;
}
@@ -2287,8 +2310,7 @@ neon_end_unlogged_build(SMgrRelation reln)
if (neon_enable_new_communicator)
{
/* TODO: we could update the cache with the size, since we have it at hand */
communicator_new_forget_cache(InfoFromSMgrRel(reln), forknum, nblocks, recptr);
communicator_new_update_cached_rel_size(InfoFromSMgrRel(reln), forknum, nblocks, recptr);
}
else
{