Line data Source code
1 : //!
2 : //! Parse PostgreSQL WAL records and store them in a neon Timeline.
3 : //!
4 : //! The pipeline for ingesting WAL looks like this:
5 : //!
6 : //! WAL receiver -> [`wal_decoder`] -> WalIngest -> Repository
7 : //!
8 : //! The WAL receiver receives a stream of WAL from the WAL safekeepers.
9 : //! Records get decoded and interpreted in the [`wal_decoder`] module
10 : //! and then stored to the Repository by WalIngest.
11 : //!
12 : //! The neon Repository can store page versions in two formats: as
13 : //! page images, or a WAL records. [`wal_decoder::models::InterpretedWalRecord::from_bytes_filtered`]
14 : //! extracts page images out of some WAL records, but mostly it's WAL
15 : //! records. If a WAL record modifies multiple pages, WalIngest
16 : //! will call Repository::put_rel_wal_record or put_rel_page_image functions
17 : //! separately for each modified page.
18 : //!
19 : //! To reconstruct a page using a WAL record, the Repository calls the
20 : //! code in walredo.rs. walredo.rs passes most WAL records to the WAL
21 : //! redo Postgres process, but some records it can handle directly with
22 : //! bespoken Rust code.
23 :
24 : use std::collections::HashMap;
25 : use std::sync::Arc;
26 : use std::sync::OnceLock;
27 : use std::time::Duration;
28 : use std::time::Instant;
29 : use std::time::SystemTime;
30 :
31 : use pageserver_api::shard::ShardIdentity;
32 : use postgres_ffi::fsm_logical_to_physical;
33 : use postgres_ffi::walrecord::*;
34 : use postgres_ffi::{dispatch_pgversion, enum_pgversion, enum_pgversion_dispatch, TimestampTz};
35 : use wal_decoder::models::*;
36 :
37 : use anyhow::{bail, Result};
38 : use bytes::{Buf, Bytes};
39 : use tracing::*;
40 : use utils::failpoint_support;
41 : use utils::rate_limit::RateLimit;
42 :
43 : use crate::context::RequestContext;
44 : use crate::metrics::WAL_INGEST;
45 : use crate::pgdatadir_mapping::{DatadirModification, Version};
46 : use crate::span::debug_assert_current_span_has_tenant_and_timeline_id;
47 : use crate::tenant::PageReconstructError;
48 : use crate::tenant::Timeline;
49 : use crate::ZERO_PAGE;
50 : use pageserver_api::key::rel_block_to_key;
51 : use pageserver_api::record::NeonWalRecord;
52 : use pageserver_api::reltag::{BlockNumber, RelTag, SlruKind};
53 : use postgres_ffi::pg_constants;
54 : use postgres_ffi::relfile_utils::{FSM_FORKNUM, INIT_FORKNUM, MAIN_FORKNUM, VISIBILITYMAP_FORKNUM};
55 : use postgres_ffi::TransactionId;
56 : use utils::bin_ser::SerializeError;
57 : use utils::lsn::Lsn;
58 :
59 : enum_pgversion! {CheckPoint, pgv::CheckPoint}
60 :
61 : impl CheckPoint {
62 6 : fn encode(&self) -> Result<Bytes, SerializeError> {
63 6 : enum_pgversion_dispatch!(self, CheckPoint, cp, { cp.encode() })
64 6 : }
65 :
66 145834 : fn update_next_xid(&mut self, xid: u32) -> bool {
67 145834 : enum_pgversion_dispatch!(self, CheckPoint, cp, { cp.update_next_xid(xid) })
68 145834 : }
69 :
70 0 : pub fn update_next_multixid(&mut self, multi_xid: u32, multi_offset: u32) -> bool {
71 0 : enum_pgversion_dispatch!(self, CheckPoint, cp, {
72 0 : cp.update_next_multixid(multi_xid, multi_offset)
73 : })
74 0 : }
75 : }
76 :
77 : /// Temporary limitation of WAL lag warnings after attach
78 : ///
79 : /// After tenant attach, we want to limit WAL lag warnings because
80 : /// we don't look at the WAL until the attach is complete, which
81 : /// might take a while.
82 : pub struct WalLagCooldown {
83 : /// Until when should this limitation apply at all
84 : active_until: std::time::Instant,
85 : /// The maximum lag to suppress. Lags above this limit get reported anyways.
86 : max_lag: Duration,
87 : }
88 :
89 : impl WalLagCooldown {
90 0 : pub fn new(attach_start: Instant, attach_duration: Duration) -> Self {
91 0 : Self {
92 0 : active_until: attach_start + attach_duration * 3 + Duration::from_secs(120),
93 0 : max_lag: attach_duration * 2 + Duration::from_secs(60),
94 0 : }
95 0 : }
96 : }
97 :
98 : pub struct WalIngest {
99 : attach_wal_lag_cooldown: Arc<OnceLock<WalLagCooldown>>,
100 : shard: ShardIdentity,
101 : checkpoint: CheckPoint,
102 : checkpoint_modified: bool,
103 : warn_ingest_lag: WarnIngestLag,
104 : }
105 :
106 : struct WarnIngestLag {
107 : lag_msg_ratelimit: RateLimit,
108 : future_lsn_msg_ratelimit: RateLimit,
109 : timestamp_invalid_msg_ratelimit: RateLimit,
110 : }
111 :
112 : impl WalIngest {
113 12 : pub async fn new(
114 12 : timeline: &Timeline,
115 12 : startpoint: Lsn,
116 12 : ctx: &RequestContext,
117 12 : ) -> anyhow::Result<WalIngest> {
118 : // Fetch the latest checkpoint into memory, so that we can compare with it
119 : // quickly in `ingest_record` and update it when it changes.
120 12 : let checkpoint_bytes = timeline.get_checkpoint(startpoint, ctx).await?;
121 12 : let pgversion = timeline.pg_version;
122 :
123 12 : let checkpoint = dispatch_pgversion!(pgversion, {
124 0 : let checkpoint = pgv::CheckPoint::decode(&checkpoint_bytes)?;
125 0 : trace!("CheckPoint.nextXid = {}", checkpoint.nextXid.value);
126 0 : <pgv::CheckPoint as Into<CheckPoint>>::into(checkpoint)
127 : });
128 :
129 12 : Ok(WalIngest {
130 12 : shard: *timeline.get_shard_identity(),
131 12 : checkpoint,
132 12 : checkpoint_modified: false,
133 12 : attach_wal_lag_cooldown: timeline.attach_wal_lag_cooldown.clone(),
134 12 : warn_ingest_lag: WarnIngestLag {
135 12 : lag_msg_ratelimit: RateLimit::new(std::time::Duration::from_secs(10)),
136 12 : future_lsn_msg_ratelimit: RateLimit::new(std::time::Duration::from_secs(10)),
137 12 : timestamp_invalid_msg_ratelimit: RateLimit::new(std::time::Duration::from_secs(10)),
138 12 : },
139 12 : })
140 12 : }
141 :
142 : /// Ingest an interpreted PostgreSQL WAL record by doing writes to the underlying key value
143 : /// storage of a given timeline.
144 : ///
145 : /// This function updates `lsn` field of `DatadirModification`
146 : ///
147 : /// This function returns `true` if the record was ingested, and `false` if it was filtered out
148 145852 : pub async fn ingest_record(
149 145852 : &mut self,
150 145852 : interpreted: InterpretedWalRecord,
151 145852 : modification: &mut DatadirModification<'_>,
152 145852 : ctx: &RequestContext,
153 145852 : ) -> anyhow::Result<bool> {
154 145852 : WAL_INGEST.records_received.inc();
155 145852 : let prev_len = modification.len();
156 145852 :
157 145852 : modification.set_lsn(interpreted.next_record_lsn)?;
158 :
159 145852 : if matches!(interpreted.flush_uncommitted, FlushUncommittedRecords::Yes) {
160 : // Records of this type should always be preceded by a commit(), as they
161 : // rely on reading data pages back from the Timeline.
162 0 : assert!(!modification.has_dirty_data());
163 145852 : }
164 :
165 145852 : assert!(!self.checkpoint_modified);
166 145852 : if interpreted.xid != pg_constants::INVALID_TRANSACTION_ID
167 145834 : && self.checkpoint.update_next_xid(interpreted.xid)
168 2 : {
169 2 : self.checkpoint_modified = true;
170 145850 : }
171 :
172 145852 : failpoint_support::sleep_millis_async!("wal-ingest-record-sleep");
173 :
174 66 : match interpreted.metadata_record {
175 12 : Some(MetadataRecord::Heapam(rec)) => match rec {
176 12 : HeapamRecord::ClearVmBits(clear_vm_bits) => {
177 12 : self.ingest_clear_vm_bits(clear_vm_bits, modification, ctx)
178 0 : .await?;
179 : }
180 : },
181 0 : Some(MetadataRecord::Neonrmgr(rec)) => match rec {
182 0 : NeonrmgrRecord::ClearVmBits(clear_vm_bits) => {
183 0 : self.ingest_clear_vm_bits(clear_vm_bits, modification, ctx)
184 0 : .await?;
185 : }
186 : },
187 16 : Some(MetadataRecord::Smgr(rec)) => match rec {
188 16 : SmgrRecord::Create(create) => {
189 16 : self.ingest_xlog_smgr_create(create, modification, ctx)
190 12 : .await?;
191 : }
192 0 : SmgrRecord::Truncate(truncate) => {
193 0 : self.ingest_xlog_smgr_truncate(truncate, modification, ctx)
194 0 : .await?;
195 : }
196 : },
197 0 : Some(MetadataRecord::Dbase(rec)) => match rec {
198 0 : DbaseRecord::Create(create) => {
199 0 : self.ingest_xlog_dbase_create(create, modification, ctx)
200 0 : .await?;
201 : }
202 0 : DbaseRecord::Drop(drop) => {
203 0 : self.ingest_xlog_dbase_drop(drop, modification, ctx).await?;
204 : }
205 : },
206 0 : Some(MetadataRecord::Clog(rec)) => match rec {
207 0 : ClogRecord::ZeroPage(zero_page) => {
208 0 : self.ingest_clog_zero_page(zero_page, modification, ctx)
209 0 : .await?;
210 : }
211 0 : ClogRecord::Truncate(truncate) => {
212 0 : self.ingest_clog_truncate(truncate, modification, ctx)
213 0 : .await?;
214 : }
215 : },
216 8 : Some(MetadataRecord::Xact(rec)) => {
217 8 : self.ingest_xact_record(rec, modification, ctx).await?;
218 : }
219 0 : Some(MetadataRecord::MultiXact(rec)) => match rec {
220 0 : MultiXactRecord::ZeroPage(zero_page) => {
221 0 : self.ingest_multixact_zero_page(zero_page, modification, ctx)
222 0 : .await?;
223 : }
224 0 : MultiXactRecord::Create(create) => {
225 0 : self.ingest_multixact_create(modification, &create)?;
226 : }
227 0 : MultiXactRecord::Truncate(truncate) => {
228 0 : self.ingest_multixact_truncate(modification, &truncate, ctx)
229 0 : .await?;
230 : }
231 : },
232 0 : Some(MetadataRecord::Relmap(rec)) => match rec {
233 0 : RelmapRecord::Update(update) => {
234 0 : self.ingest_relmap_update(update, modification, ctx).await?;
235 : }
236 : },
237 30 : Some(MetadataRecord::Xlog(rec)) => match rec {
238 30 : XlogRecord::Raw(raw) => {
239 30 : self.ingest_raw_xlog_record(raw, modification, ctx).await?;
240 : }
241 : },
242 0 : Some(MetadataRecord::LogicalMessage(rec)) => match rec {
243 0 : LogicalMessageRecord::Put(put) => {
244 0 : self.ingest_logical_message_put(put, modification, ctx)
245 0 : .await?;
246 : }
247 : #[cfg(feature = "testing")]
248 : LogicalMessageRecord::Failpoint => {
249 : // This is a convenient way to make the WAL ingestion pause at
250 : // particular point in the WAL. For more fine-grained control,
251 : // we could peek into the message and only pause if it contains
252 : // a particular string, for example, but this is enough for now.
253 0 : failpoint_support::sleep_millis_async!(
254 0 : "pageserver-wal-ingest-logical-message-sleep"
255 0 : );
256 : }
257 : },
258 0 : Some(MetadataRecord::Standby(rec)) => {
259 0 : self.ingest_standby_record(rec).unwrap();
260 0 : }
261 0 : Some(MetadataRecord::Replorigin(rec)) => {
262 0 : self.ingest_replorigin_record(rec, modification).await?;
263 : }
264 145786 : None => {
265 145786 : // There are two cases through which we end up here:
266 145786 : // 1. The resource manager for the original PG WAL record
267 145786 : // is [`pg_constants::RM_TBLSPC_ID`]. This is not a supported
268 145786 : // record type within Neon.
269 145786 : // 2. The resource manager id was unknown to
270 145786 : // [`wal_decoder::decoder::MetadataRecord::from_decoded`].
271 145786 : // TODO(vlad): Tighten this up more once we build confidence
272 145786 : // that case (2) does not happen in the field.
273 145786 : }
274 : }
275 :
276 145852 : modification
277 145852 : .ingest_batch(interpreted.batch, &self.shard, ctx)
278 284 : .await?;
279 :
280 : // If checkpoint data was updated, store the new version in the repository
281 145852 : if self.checkpoint_modified {
282 6 : let new_checkpoint_bytes = self.checkpoint.encode()?;
283 :
284 6 : modification.put_checkpoint(new_checkpoint_bytes)?;
285 6 : self.checkpoint_modified = false;
286 145846 : }
287 :
288 : // Note that at this point this record is only cached in the modification
289 : // until commit() is called to flush the data into the repository and update
290 : // the latest LSN.
291 :
292 145852 : Ok(modification.len() > prev_len)
293 145852 : }
294 :
295 : /// This is the same as AdjustToFullTransactionId(xid) in PostgreSQL
296 0 : fn adjust_to_full_transaction_id(&self, xid: TransactionId) -> Result<u64> {
297 0 : let next_full_xid =
298 0 : enum_pgversion_dispatch!(&self.checkpoint, CheckPoint, cp, { cp.nextXid.value });
299 :
300 0 : let next_xid = (next_full_xid) as u32;
301 0 : let mut epoch = (next_full_xid >> 32) as u32;
302 0 :
303 0 : if xid > next_xid {
304 : // Wraparound occurred, must be from a prev epoch.
305 0 : if epoch == 0 {
306 0 : bail!("apparent XID wraparound with prepared transaction XID {xid}, nextXid is {next_full_xid}");
307 0 : }
308 0 : epoch -= 1;
309 0 : }
310 :
311 0 : Ok((epoch as u64) << 32 | xid as u64)
312 0 : }
313 :
314 12 : async fn ingest_clear_vm_bits(
315 12 : &mut self,
316 12 : clear_vm_bits: ClearVmBits,
317 12 : modification: &mut DatadirModification<'_>,
318 12 : ctx: &RequestContext,
319 12 : ) -> anyhow::Result<()> {
320 12 : let ClearVmBits {
321 12 : new_heap_blkno,
322 12 : old_heap_blkno,
323 12 : flags,
324 12 : vm_rel,
325 12 : } = clear_vm_bits;
326 12 : // Clear the VM bits if required.
327 12 : let mut new_vm_blk = new_heap_blkno.map(pg_constants::HEAPBLK_TO_MAPBLOCK);
328 12 : let mut old_vm_blk = old_heap_blkno.map(pg_constants::HEAPBLK_TO_MAPBLOCK);
329 :
330 : // Sometimes, Postgres seems to create heap WAL records with the
331 : // ALL_VISIBLE_CLEARED flag set, even though the bit in the VM page is
332 : // not set. In fact, it's possible that the VM page does not exist at all.
333 : // In that case, we don't want to store a record to clear the VM bit;
334 : // replaying it would fail to find the previous image of the page, because
335 : // it doesn't exist. So check if the VM page(s) exist, and skip the WAL
336 : // record if it doesn't.
337 12 : let vm_size = get_relsize(modification, vm_rel, ctx).await?;
338 12 : if let Some(blknum) = new_vm_blk {
339 12 : if blknum >= vm_size {
340 0 : new_vm_blk = None;
341 12 : }
342 0 : }
343 12 : if let Some(blknum) = old_vm_blk {
344 0 : if blknum >= vm_size {
345 0 : old_vm_blk = None;
346 0 : }
347 12 : }
348 :
349 12 : if new_vm_blk.is_some() || old_vm_blk.is_some() {
350 12 : if new_vm_blk == old_vm_blk {
351 : // An UPDATE record that needs to clear the bits for both old and the
352 : // new page, both of which reside on the same VM page.
353 0 : self.put_rel_wal_record(
354 0 : modification,
355 0 : vm_rel,
356 0 : new_vm_blk.unwrap(),
357 0 : NeonWalRecord::ClearVisibilityMapFlags {
358 0 : new_heap_blkno,
359 0 : old_heap_blkno,
360 0 : flags,
361 0 : },
362 0 : ctx,
363 0 : )
364 0 : .await?;
365 : } else {
366 : // Clear VM bits for one heap page, or for two pages that reside on
367 : // different VM pages.
368 12 : if let Some(new_vm_blk) = new_vm_blk {
369 12 : self.put_rel_wal_record(
370 12 : modification,
371 12 : vm_rel,
372 12 : new_vm_blk,
373 12 : NeonWalRecord::ClearVisibilityMapFlags {
374 12 : new_heap_blkno,
375 12 : old_heap_blkno: None,
376 12 : flags,
377 12 : },
378 12 : ctx,
379 12 : )
380 0 : .await?;
381 0 : }
382 12 : if let Some(old_vm_blk) = old_vm_blk {
383 0 : self.put_rel_wal_record(
384 0 : modification,
385 0 : vm_rel,
386 0 : old_vm_blk,
387 0 : NeonWalRecord::ClearVisibilityMapFlags {
388 0 : new_heap_blkno: None,
389 0 : old_heap_blkno,
390 0 : flags,
391 0 : },
392 0 : ctx,
393 0 : )
394 0 : .await?;
395 12 : }
396 : }
397 0 : }
398 :
399 12 : Ok(())
400 12 : }
401 :
402 : /// Subroutine of ingest_record(), to handle an XLOG_DBASE_CREATE record.
403 0 : async fn ingest_xlog_dbase_create(
404 0 : &mut self,
405 0 : create: DbaseCreate,
406 0 : modification: &mut DatadirModification<'_>,
407 0 : ctx: &RequestContext,
408 0 : ) -> anyhow::Result<()> {
409 0 : let DbaseCreate {
410 0 : db_id,
411 0 : tablespace_id,
412 0 : src_db_id,
413 0 : src_tablespace_id,
414 0 : } = create;
415 :
416 0 : let rels = modification
417 0 : .tline
418 0 : .list_rels(
419 0 : src_tablespace_id,
420 0 : src_db_id,
421 0 : Version::Modified(modification),
422 0 : ctx,
423 0 : )
424 0 : .await?;
425 :
426 0 : debug!("ingest_xlog_dbase_create: {} rels", rels.len());
427 :
428 : // Copy relfilemap
429 0 : let filemap = modification
430 0 : .tline
431 0 : .get_relmap_file(
432 0 : src_tablespace_id,
433 0 : src_db_id,
434 0 : Version::Modified(modification),
435 0 : ctx,
436 0 : )
437 0 : .await?;
438 0 : modification
439 0 : .put_relmap_file(tablespace_id, db_id, filemap, ctx)
440 0 : .await?;
441 :
442 0 : let mut num_rels_copied = 0;
443 0 : let mut num_blocks_copied = 0;
444 0 : for src_rel in rels {
445 0 : assert_eq!(src_rel.spcnode, src_tablespace_id);
446 0 : assert_eq!(src_rel.dbnode, src_db_id);
447 :
448 0 : let nblocks = modification
449 0 : .tline
450 0 : .get_rel_size(src_rel, Version::Modified(modification), ctx)
451 0 : .await?;
452 0 : let dst_rel = RelTag {
453 0 : spcnode: tablespace_id,
454 0 : dbnode: db_id,
455 0 : relnode: src_rel.relnode,
456 0 : forknum: src_rel.forknum,
457 0 : };
458 0 :
459 0 : modification.put_rel_creation(dst_rel, nblocks, ctx).await?;
460 :
461 : // Copy content
462 0 : debug!("copying rel {} to {}, {} blocks", src_rel, dst_rel, nblocks);
463 0 : for blknum in 0..nblocks {
464 : // Sharding:
465 : // - src and dst are always on the same shard, because they differ only by dbNode, and
466 : // dbNode is not included in the hash inputs for sharding.
467 : // - This WAL command is replayed on all shards, but each shard only copies the blocks
468 : // that belong to it.
469 0 : let src_key = rel_block_to_key(src_rel, blknum);
470 0 : if !self.shard.is_key_local(&src_key) {
471 0 : debug!(
472 0 : "Skipping non-local key {} during XLOG_DBASE_CREATE",
473 : src_key
474 : );
475 0 : continue;
476 0 : }
477 0 : debug!(
478 0 : "copying block {} from {} ({}) to {}",
479 : blknum, src_rel, src_key, dst_rel
480 : );
481 :
482 0 : let content = modification
483 0 : .tline
484 0 : .get_rel_page_at_lsn(src_rel, blknum, Version::Modified(modification), ctx)
485 0 : .await?;
486 0 : modification.put_rel_page_image(dst_rel, blknum, content)?;
487 0 : num_blocks_copied += 1;
488 : }
489 :
490 0 : num_rels_copied += 1;
491 : }
492 :
493 0 : info!(
494 0 : "Created database {}/{}, copied {} blocks in {} rels",
495 : tablespace_id, db_id, num_blocks_copied, num_rels_copied
496 : );
497 0 : Ok(())
498 0 : }
499 :
500 0 : async fn ingest_xlog_dbase_drop(
501 0 : &mut self,
502 0 : dbase_drop: DbaseDrop,
503 0 : modification: &mut DatadirModification<'_>,
504 0 : ctx: &RequestContext,
505 0 : ) -> anyhow::Result<()> {
506 0 : let DbaseDrop {
507 0 : db_id,
508 0 : tablespace_ids,
509 0 : } = dbase_drop;
510 0 : for tablespace_id in tablespace_ids {
511 0 : trace!("Drop db {}, {}", tablespace_id, db_id);
512 0 : modification.drop_dbdir(tablespace_id, db_id, ctx).await?;
513 : }
514 :
515 0 : Ok(())
516 0 : }
517 :
518 16 : async fn ingest_xlog_smgr_create(
519 16 : &mut self,
520 16 : create: SmgrCreate,
521 16 : modification: &mut DatadirModification<'_>,
522 16 : ctx: &RequestContext,
523 16 : ) -> anyhow::Result<()> {
524 16 : let SmgrCreate { rel } = create;
525 16 : self.put_rel_creation(modification, rel, ctx).await?;
526 16 : Ok(())
527 16 : }
528 :
529 : /// Subroutine of ingest_record(), to handle an XLOG_SMGR_TRUNCATE record.
530 : ///
531 : /// This is the same logic as in PostgreSQL's smgr_redo() function.
532 0 : async fn ingest_xlog_smgr_truncate(
533 0 : &mut self,
534 0 : truncate: XlSmgrTruncate,
535 0 : modification: &mut DatadirModification<'_>,
536 0 : ctx: &RequestContext,
537 0 : ) -> anyhow::Result<()> {
538 0 : let XlSmgrTruncate {
539 0 : blkno,
540 0 : rnode,
541 0 : flags,
542 0 : } = truncate;
543 0 :
544 0 : let spcnode = rnode.spcnode;
545 0 : let dbnode = rnode.dbnode;
546 0 : let relnode = rnode.relnode;
547 0 :
548 0 : if flags & pg_constants::SMGR_TRUNCATE_HEAP != 0 {
549 0 : let rel = RelTag {
550 0 : spcnode,
551 0 : dbnode,
552 0 : relnode,
553 0 : forknum: MAIN_FORKNUM,
554 0 : };
555 0 :
556 0 : self.put_rel_truncation(modification, rel, blkno, ctx)
557 0 : .await?;
558 0 : }
559 0 : if flags & pg_constants::SMGR_TRUNCATE_FSM != 0 {
560 0 : let rel = RelTag {
561 0 : spcnode,
562 0 : dbnode,
563 0 : relnode,
564 0 : forknum: FSM_FORKNUM,
565 0 : };
566 0 :
567 0 : let fsm_logical_page_no = blkno / pg_constants::SLOTS_PER_FSM_PAGE;
568 0 : let mut fsm_physical_page_no = fsm_logical_to_physical(fsm_logical_page_no);
569 0 : if blkno % pg_constants::SLOTS_PER_FSM_PAGE != 0 {
570 : // Tail of last remaining FSM page has to be zeroed.
571 : // We are not precise here and instead of digging in FSM bitmap format just clear the whole page.
572 0 : modification.put_rel_page_image_zero(rel, fsm_physical_page_no)?;
573 0 : fsm_physical_page_no += 1;
574 0 : }
575 0 : let nblocks = get_relsize(modification, rel, ctx).await?;
576 0 : if nblocks > fsm_physical_page_no {
577 : // check if something to do: FSM is larger than truncate position
578 0 : self.put_rel_truncation(modification, rel, fsm_physical_page_no, ctx)
579 0 : .await?;
580 0 : }
581 0 : }
582 0 : if flags & pg_constants::SMGR_TRUNCATE_VM != 0 {
583 0 : let rel = RelTag {
584 0 : spcnode,
585 0 : dbnode,
586 0 : relnode,
587 0 : forknum: VISIBILITYMAP_FORKNUM,
588 0 : };
589 0 :
590 0 : // last remaining block, byte, and bit
591 0 : let mut vm_page_no = blkno / (pg_constants::VM_HEAPBLOCKS_PER_PAGE as u32);
592 0 : let trunc_byte = blkno as usize % pg_constants::VM_HEAPBLOCKS_PER_PAGE
593 0 : / pg_constants::VM_HEAPBLOCKS_PER_BYTE;
594 0 : let trunc_offs = blkno as usize % pg_constants::VM_HEAPBLOCKS_PER_BYTE
595 0 : * pg_constants::VM_BITS_PER_HEAPBLOCK;
596 0 :
597 0 : // Unless the new size is exactly at a visibility map page boundary, the
598 0 : // tail bits in the last remaining map page, representing truncated heap
599 0 : // blocks, need to be cleared. This is not only tidy, but also necessary
600 0 : // because we don't get a chance to clear the bits if the heap is extended
601 0 : // again.
602 0 : if (trunc_byte != 0 || trunc_offs != 0)
603 0 : && self.shard.is_key_local(&rel_block_to_key(rel, vm_page_no))
604 : {
605 0 : modification.put_rel_wal_record(
606 0 : rel,
607 0 : vm_page_no,
608 0 : NeonWalRecord::TruncateVisibilityMap {
609 0 : trunc_byte,
610 0 : trunc_offs,
611 0 : },
612 0 : )?;
613 0 : vm_page_no += 1;
614 0 : }
615 0 : let nblocks = get_relsize(modification, rel, ctx).await?;
616 0 : if nblocks > vm_page_no {
617 : // check if something to do: VM is larger than truncate position
618 0 : self.put_rel_truncation(modification, rel, vm_page_no, ctx)
619 0 : .await?;
620 0 : }
621 0 : }
622 0 : Ok(())
623 0 : }
624 :
625 8 : fn warn_on_ingest_lag(
626 8 : &mut self,
627 8 : conf: &crate::config::PageServerConf,
628 8 : wal_timestamp: TimestampTz,
629 8 : ) {
630 8 : debug_assert_current_span_has_tenant_and_timeline_id();
631 8 : let now = SystemTime::now();
632 8 : let rate_limits = &mut self.warn_ingest_lag;
633 :
634 8 : let ts = enum_pgversion_dispatch!(&self.checkpoint, CheckPoint, _cp, {
635 0 : pgv::xlog_utils::try_from_pg_timestamp(wal_timestamp)
636 : });
637 :
638 8 : match ts {
639 8 : Ok(ts) => {
640 8 : match now.duration_since(ts) {
641 8 : Ok(lag) => {
642 8 : if lag > conf.wait_lsn_timeout {
643 8 : rate_limits.lag_msg_ratelimit.call2(|rate_limit_stats| {
644 2 : if let Some(cooldown) = self.attach_wal_lag_cooldown.get() {
645 0 : if std::time::Instant::now() < cooldown.active_until && lag <= cooldown.max_lag {
646 0 : return;
647 0 : }
648 2 : } else {
649 2 : // Still loading? We shouldn't be here
650 2 : }
651 2 : let lag = humantime::format_duration(lag);
652 2 : warn!(%rate_limit_stats, %lag, "ingesting record with timestamp lagging more than wait_lsn_timeout");
653 8 : })
654 0 : }
655 : }
656 0 : Err(e) => {
657 0 : let delta_t = e.duration();
658 : // determined by prod victoriametrics query: 1000 * (timestamp(node_time_seconds{neon_service="pageserver"}) - node_time_seconds)
659 : // => https://www.robustperception.io/time-metric-from-the-node-exporter/
660 : const IGNORED_DRIFT: Duration = Duration::from_millis(100);
661 0 : if delta_t > IGNORED_DRIFT {
662 0 : let delta_t = humantime::format_duration(delta_t);
663 0 : rate_limits.future_lsn_msg_ratelimit.call2(|rate_limit_stats| {
664 0 : warn!(%rate_limit_stats, %delta_t, "ingesting record with timestamp from future");
665 0 : })
666 0 : }
667 : }
668 : };
669 : }
670 0 : Err(error) => {
671 0 : rate_limits.timestamp_invalid_msg_ratelimit.call2(|rate_limit_stats| {
672 0 : warn!(%rate_limit_stats, %error, "ingesting record with invalid timestamp, cannot calculate lag and will fail find-lsn-for-timestamp type queries");
673 0 : })
674 : }
675 : }
676 8 : }
677 :
678 : /// Subroutine of ingest_record(), to handle an XLOG_XACT_* records.
679 : ///
680 8 : async fn ingest_xact_record(
681 8 : &mut self,
682 8 : record: XactRecord,
683 8 : modification: &mut DatadirModification<'_>,
684 8 : ctx: &RequestContext,
685 8 : ) -> anyhow::Result<()> {
686 8 : let (xact_common, is_commit, is_prepared) = match record {
687 0 : XactRecord::Prepare(XactPrepare { xl_xid, data }) => {
688 0 : let xid: u64 = if modification.tline.pg_version >= 17 {
689 0 : self.adjust_to_full_transaction_id(xl_xid)?
690 : } else {
691 0 : xl_xid as u64
692 : };
693 0 : return modification.put_twophase_file(xid, data, ctx).await;
694 : }
695 8 : XactRecord::Commit(common) => (common, true, false),
696 0 : XactRecord::Abort(common) => (common, false, false),
697 0 : XactRecord::CommitPrepared(common) => (common, true, true),
698 0 : XactRecord::AbortPrepared(common) => (common, false, true),
699 : };
700 :
701 : let XactCommon {
702 8 : parsed,
703 8 : origin_id,
704 8 : xl_xid,
705 8 : lsn,
706 8 : } = xact_common;
707 8 :
708 8 : // Record update of CLOG pages
709 8 : let mut pageno = parsed.xid / pg_constants::CLOG_XACTS_PER_PAGE;
710 8 : let mut segno = pageno / pg_constants::SLRU_PAGES_PER_SEGMENT;
711 8 : let mut rpageno = pageno % pg_constants::SLRU_PAGES_PER_SEGMENT;
712 8 : let mut page_xids: Vec<TransactionId> = vec![parsed.xid];
713 8 :
714 8 : self.warn_on_ingest_lag(modification.tline.conf, parsed.xact_time);
715 :
716 8 : for subxact in &parsed.subxacts {
717 0 : let subxact_pageno = subxact / pg_constants::CLOG_XACTS_PER_PAGE;
718 0 : if subxact_pageno != pageno {
719 : // This subxact goes to different page. Write the record
720 : // for all the XIDs on the previous page, and continue
721 : // accumulating XIDs on this new page.
722 0 : modification.put_slru_wal_record(
723 0 : SlruKind::Clog,
724 0 : segno,
725 0 : rpageno,
726 0 : if is_commit {
727 0 : NeonWalRecord::ClogSetCommitted {
728 0 : xids: page_xids,
729 0 : timestamp: parsed.xact_time,
730 0 : }
731 : } else {
732 0 : NeonWalRecord::ClogSetAborted { xids: page_xids }
733 : },
734 0 : )?;
735 0 : page_xids = Vec::new();
736 0 : }
737 0 : pageno = subxact_pageno;
738 0 : segno = pageno / pg_constants::SLRU_PAGES_PER_SEGMENT;
739 0 : rpageno = pageno % pg_constants::SLRU_PAGES_PER_SEGMENT;
740 0 : page_xids.push(*subxact);
741 : }
742 8 : modification.put_slru_wal_record(
743 8 : SlruKind::Clog,
744 8 : segno,
745 8 : rpageno,
746 8 : if is_commit {
747 8 : NeonWalRecord::ClogSetCommitted {
748 8 : xids: page_xids,
749 8 : timestamp: parsed.xact_time,
750 8 : }
751 : } else {
752 0 : NeonWalRecord::ClogSetAborted { xids: page_xids }
753 : },
754 0 : )?;
755 :
756 : // Group relations to drop by dbNode. This map will contain all relations that _might_
757 : // exist, we will reduce it to which ones really exist later. This map can be huge if
758 : // the transaction touches a huge number of relations (there is no bound on this in
759 : // postgres).
760 8 : let mut drop_relations: HashMap<(u32, u32), Vec<RelTag>> = HashMap::new();
761 :
762 8 : for xnode in &parsed.xnodes {
763 0 : for forknum in MAIN_FORKNUM..=INIT_FORKNUM {
764 0 : let rel = RelTag {
765 0 : forknum,
766 0 : spcnode: xnode.spcnode,
767 0 : dbnode: xnode.dbnode,
768 0 : relnode: xnode.relnode,
769 0 : };
770 0 : drop_relations
771 0 : .entry((xnode.spcnode, xnode.dbnode))
772 0 : .or_default()
773 0 : .push(rel);
774 0 : }
775 : }
776 :
777 : // Execute relation drops in a batch: the number may be huge, so deleting individually is prohibitively expensive
778 8 : modification.put_rel_drops(drop_relations, ctx).await?;
779 :
780 8 : if origin_id != 0 {
781 0 : modification
782 0 : .set_replorigin(origin_id, parsed.origin_lsn)
783 0 : .await?;
784 8 : }
785 :
786 8 : if is_prepared {
787 : // Remove twophase file. see RemoveTwoPhaseFile() in postgres code
788 0 : trace!(
789 0 : "Drop twophaseFile for xid {} parsed_xact.xid {} here at {}",
790 : xl_xid,
791 : parsed.xid,
792 : lsn,
793 : );
794 :
795 0 : let xid: u64 = if modification.tline.pg_version >= 17 {
796 0 : self.adjust_to_full_transaction_id(parsed.xid)?
797 : } else {
798 0 : parsed.xid as u64
799 : };
800 0 : modification.drop_twophase_file(xid, ctx).await?;
801 8 : }
802 :
803 8 : Ok(())
804 8 : }
805 :
806 0 : async fn ingest_clog_truncate(
807 0 : &mut self,
808 0 : truncate: ClogTruncate,
809 0 : modification: &mut DatadirModification<'_>,
810 0 : ctx: &RequestContext,
811 0 : ) -> anyhow::Result<()> {
812 0 : let ClogTruncate {
813 0 : pageno,
814 0 : oldest_xid,
815 0 : oldest_xid_db,
816 0 : } = truncate;
817 0 :
818 0 : info!(
819 0 : "RM_CLOG_ID truncate pageno {} oldestXid {} oldestXidDB {}",
820 : pageno, oldest_xid, oldest_xid_db
821 : );
822 :
823 : // In Postgres, oldestXid and oldestXidDB are updated in memory when the CLOG is
824 : // truncated, but a checkpoint record with the updated values isn't written until
825 : // later. In Neon, a server can start at any LSN, not just on a checkpoint record,
826 : // so we keep the oldestXid and oldestXidDB up-to-date.
827 0 : enum_pgversion_dispatch!(&mut self.checkpoint, CheckPoint, cp, {
828 0 : cp.oldestXid = oldest_xid;
829 0 : cp.oldestXidDB = oldest_xid_db;
830 0 : });
831 0 : self.checkpoint_modified = true;
832 :
833 : // TODO Treat AdvanceOldestClogXid() or write a comment why we don't need it
834 :
835 0 : let latest_page_number =
836 0 : enum_pgversion_dispatch!(self.checkpoint, CheckPoint, cp, { cp.nextXid.value }) as u32
837 : / pg_constants::CLOG_XACTS_PER_PAGE;
838 :
839 : // Now delete all segments containing pages between xlrec.pageno
840 : // and latest_page_number.
841 :
842 : // First, make an important safety check:
843 : // the current endpoint page must not be eligible for removal.
844 : // See SimpleLruTruncate() in slru.c
845 0 : if dispatch_pgversion!(modification.tline.pg_version, {
846 0 : pgv::nonrelfile_utils::clogpage_precedes(latest_page_number, pageno)
847 : }) {
848 0 : info!("could not truncate directory pg_xact apparent wraparound");
849 0 : return Ok(());
850 0 : }
851 :
852 : // Iterate via SLRU CLOG segments and drop segments that we're ready to truncate
853 : //
854 : // We cannot pass 'lsn' to the Timeline.list_nonrels(), or it
855 : // will block waiting for the last valid LSN to advance up to
856 : // it. So we use the previous record's LSN in the get calls
857 : // instead.
858 0 : for segno in modification
859 0 : .tline
860 0 : .list_slru_segments(SlruKind::Clog, Version::Modified(modification), ctx)
861 0 : .await?
862 : {
863 0 : let segpage = segno * pg_constants::SLRU_PAGES_PER_SEGMENT;
864 :
865 0 : let may_delete = dispatch_pgversion!(modification.tline.pg_version, {
866 0 : pgv::nonrelfile_utils::slru_may_delete_clogsegment(segpage, pageno)
867 : });
868 :
869 0 : if may_delete {
870 0 : modification
871 0 : .drop_slru_segment(SlruKind::Clog, segno, ctx)
872 0 : .await?;
873 0 : trace!("Drop CLOG segment {:>04X}", segno);
874 0 : }
875 : }
876 :
877 0 : Ok(())
878 0 : }
879 :
880 0 : async fn ingest_clog_zero_page(
881 0 : &mut self,
882 0 : zero_page: ClogZeroPage,
883 0 : modification: &mut DatadirModification<'_>,
884 0 : ctx: &RequestContext,
885 0 : ) -> anyhow::Result<()> {
886 0 : let ClogZeroPage { segno, rpageno } = zero_page;
887 0 :
888 0 : self.put_slru_page_image(
889 0 : modification,
890 0 : SlruKind::Clog,
891 0 : segno,
892 0 : rpageno,
893 0 : ZERO_PAGE.clone(),
894 0 : ctx,
895 0 : )
896 0 : .await
897 0 : }
898 :
899 0 : fn ingest_multixact_create(
900 0 : &mut self,
901 0 : modification: &mut DatadirModification,
902 0 : xlrec: &XlMultiXactCreate,
903 0 : ) -> Result<()> {
904 0 : // Create WAL record for updating the multixact-offsets page
905 0 : let pageno = xlrec.mid / pg_constants::MULTIXACT_OFFSETS_PER_PAGE as u32;
906 0 : let segno = pageno / pg_constants::SLRU_PAGES_PER_SEGMENT;
907 0 : let rpageno = pageno % pg_constants::SLRU_PAGES_PER_SEGMENT;
908 0 :
909 0 : modification.put_slru_wal_record(
910 0 : SlruKind::MultiXactOffsets,
911 0 : segno,
912 0 : rpageno,
913 0 : NeonWalRecord::MultixactOffsetCreate {
914 0 : mid: xlrec.mid,
915 0 : moff: xlrec.moff,
916 0 : },
917 0 : )?;
918 :
919 : // Create WAL records for the update of each affected multixact-members page
920 0 : let mut members = xlrec.members.iter();
921 0 : let mut offset = xlrec.moff;
922 : loop {
923 0 : let pageno = offset / pg_constants::MULTIXACT_MEMBERS_PER_PAGE as u32;
924 0 :
925 0 : // How many members fit on this page?
926 0 : let page_remain = pg_constants::MULTIXACT_MEMBERS_PER_PAGE as u32
927 0 : - offset % pg_constants::MULTIXACT_MEMBERS_PER_PAGE as u32;
928 0 :
929 0 : let mut this_page_members: Vec<MultiXactMember> = Vec::new();
930 0 : for _ in 0..page_remain {
931 0 : if let Some(m) = members.next() {
932 0 : this_page_members.push(m.clone());
933 0 : } else {
934 0 : break;
935 : }
936 : }
937 0 : if this_page_members.is_empty() {
938 : // all done
939 0 : break;
940 0 : }
941 0 : let n_this_page = this_page_members.len();
942 0 :
943 0 : modification.put_slru_wal_record(
944 0 : SlruKind::MultiXactMembers,
945 0 : pageno / pg_constants::SLRU_PAGES_PER_SEGMENT,
946 0 : pageno % pg_constants::SLRU_PAGES_PER_SEGMENT,
947 0 : NeonWalRecord::MultixactMembersCreate {
948 0 : moff: offset,
949 0 : members: this_page_members,
950 0 : },
951 0 : )?;
952 :
953 : // Note: The multixact members can wrap around, even within one WAL record.
954 0 : offset = offset.wrapping_add(n_this_page as u32);
955 : }
956 0 : let next_offset = offset;
957 0 : assert!(xlrec.moff.wrapping_add(xlrec.nmembers) == next_offset);
958 :
959 : // Update next-multi-xid and next-offset
960 : //
961 : // NB: In PostgreSQL, the next-multi-xid stored in the control file is allowed to
962 : // go to 0, and it's fixed up by skipping to FirstMultiXactId in functions that
963 : // read it, like GetNewMultiXactId(). This is different from how nextXid is
964 : // incremented! nextXid skips over < FirstNormalTransactionId when the the value
965 : // is stored, so it's never 0 in a checkpoint.
966 : //
967 : // I don't know why it's done that way, it seems less error-prone to skip over 0
968 : // when the value is stored rather than when it's read. But let's do it the same
969 : // way here.
970 0 : let next_multi_xid = xlrec.mid.wrapping_add(1);
971 0 :
972 0 : if self
973 0 : .checkpoint
974 0 : .update_next_multixid(next_multi_xid, next_offset)
975 0 : {
976 0 : self.checkpoint_modified = true;
977 0 : }
978 :
979 : // Also update the next-xid with the highest member. According to the comments in
980 : // multixact_redo(), this shouldn't be necessary, but let's do the same here.
981 0 : let max_mbr_xid = xlrec.members.iter().fold(None, |acc, mbr| {
982 0 : if let Some(max_xid) = acc {
983 0 : if mbr.xid.wrapping_sub(max_xid) as i32 > 0 {
984 0 : Some(mbr.xid)
985 : } else {
986 0 : acc
987 : }
988 : } else {
989 0 : Some(mbr.xid)
990 : }
991 0 : });
992 :
993 0 : if let Some(max_xid) = max_mbr_xid {
994 0 : if self.checkpoint.update_next_xid(max_xid) {
995 0 : self.checkpoint_modified = true;
996 0 : }
997 0 : }
998 0 : Ok(())
999 0 : }
1000 :
1001 0 : async fn ingest_multixact_truncate(
1002 0 : &mut self,
1003 0 : modification: &mut DatadirModification<'_>,
1004 0 : xlrec: &XlMultiXactTruncate,
1005 0 : ctx: &RequestContext,
1006 0 : ) -> Result<()> {
1007 0 : let (maxsegment, startsegment, endsegment) =
1008 0 : enum_pgversion_dispatch!(&mut self.checkpoint, CheckPoint, cp, {
1009 0 : cp.oldestMulti = xlrec.end_trunc_off;
1010 0 : cp.oldestMultiDB = xlrec.oldest_multi_db;
1011 0 : let maxsegment: i32 = pgv::nonrelfile_utils::mx_offset_to_member_segment(
1012 0 : pg_constants::MAX_MULTIXACT_OFFSET,
1013 0 : );
1014 0 : let startsegment: i32 =
1015 0 : pgv::nonrelfile_utils::mx_offset_to_member_segment(xlrec.start_trunc_memb);
1016 0 : let endsegment: i32 =
1017 0 : pgv::nonrelfile_utils::mx_offset_to_member_segment(xlrec.end_trunc_memb);
1018 0 : (maxsegment, startsegment, endsegment)
1019 : });
1020 :
1021 0 : self.checkpoint_modified = true;
1022 0 :
1023 0 : // PerformMembersTruncation
1024 0 : let mut segment: i32 = startsegment;
1025 :
1026 : // Delete all the segments except the last one. The last segment can still
1027 : // contain, possibly partially, valid data.
1028 0 : while segment != endsegment {
1029 0 : modification
1030 0 : .drop_slru_segment(SlruKind::MultiXactMembers, segment as u32, ctx)
1031 0 : .await?;
1032 :
1033 : /* move to next segment, handling wraparound correctly */
1034 0 : if segment == maxsegment {
1035 0 : segment = 0;
1036 0 : } else {
1037 0 : segment += 1;
1038 0 : }
1039 : }
1040 :
1041 : // Truncate offsets
1042 : // FIXME: this did not handle wraparound correctly
1043 :
1044 0 : Ok(())
1045 0 : }
1046 :
1047 0 : async fn ingest_multixact_zero_page(
1048 0 : &mut self,
1049 0 : zero_page: MultiXactZeroPage,
1050 0 : modification: &mut DatadirModification<'_>,
1051 0 : ctx: &RequestContext,
1052 0 : ) -> Result<()> {
1053 0 : let MultiXactZeroPage {
1054 0 : slru_kind,
1055 0 : segno,
1056 0 : rpageno,
1057 0 : } = zero_page;
1058 0 : self.put_slru_page_image(
1059 0 : modification,
1060 0 : slru_kind,
1061 0 : segno,
1062 0 : rpageno,
1063 0 : ZERO_PAGE.clone(),
1064 0 : ctx,
1065 0 : )
1066 0 : .await
1067 0 : }
1068 :
1069 0 : async fn ingest_relmap_update(
1070 0 : &mut self,
1071 0 : update: RelmapUpdate,
1072 0 : modification: &mut DatadirModification<'_>,
1073 0 : ctx: &RequestContext,
1074 0 : ) -> Result<()> {
1075 0 : let RelmapUpdate { update, buf } = update;
1076 0 :
1077 0 : modification
1078 0 : .put_relmap_file(update.tsid, update.dbid, buf, ctx)
1079 0 : .await
1080 0 : }
1081 :
1082 30 : async fn ingest_raw_xlog_record(
1083 30 : &mut self,
1084 30 : raw_record: RawXlogRecord,
1085 30 : modification: &mut DatadirModification<'_>,
1086 30 : ctx: &RequestContext,
1087 30 : ) -> Result<()> {
1088 30 : let RawXlogRecord { info, lsn, mut buf } = raw_record;
1089 30 : let pg_version = modification.tline.pg_version;
1090 30 :
1091 30 : if info == pg_constants::XLOG_PARAMETER_CHANGE {
1092 2 : if let CheckPoint::V17(cp) = &mut self.checkpoint {
1093 0 : let rec = v17::XlParameterChange::decode(&mut buf);
1094 0 : cp.wal_level = rec.wal_level;
1095 0 : self.checkpoint_modified = true;
1096 2 : }
1097 28 : } else if info == pg_constants::XLOG_END_OF_RECOVERY {
1098 0 : if let CheckPoint::V17(cp) = &mut self.checkpoint {
1099 0 : let rec = v17::XlEndOfRecovery::decode(&mut buf);
1100 0 : cp.wal_level = rec.wal_level;
1101 0 : self.checkpoint_modified = true;
1102 0 : }
1103 28 : }
1104 :
1105 30 : enum_pgversion_dispatch!(&mut self.checkpoint, CheckPoint, cp, {
1106 0 : if info == pg_constants::XLOG_NEXTOID {
1107 0 : let next_oid = buf.get_u32_le();
1108 0 : if cp.nextOid != next_oid {
1109 0 : cp.nextOid = next_oid;
1110 0 : self.checkpoint_modified = true;
1111 0 : }
1112 0 : } else if info == pg_constants::XLOG_CHECKPOINT_ONLINE
1113 0 : || info == pg_constants::XLOG_CHECKPOINT_SHUTDOWN
1114 : {
1115 0 : let mut checkpoint_bytes = [0u8; pgv::xlog_utils::SIZEOF_CHECKPOINT];
1116 0 : buf.copy_to_slice(&mut checkpoint_bytes);
1117 0 : let xlog_checkpoint = pgv::CheckPoint::decode(&checkpoint_bytes)?;
1118 0 : trace!(
1119 0 : "xlog_checkpoint.oldestXid={}, checkpoint.oldestXid={}",
1120 : xlog_checkpoint.oldestXid,
1121 : cp.oldestXid
1122 : );
1123 0 : if (cp.oldestXid.wrapping_sub(xlog_checkpoint.oldestXid) as i32) < 0 {
1124 0 : cp.oldestXid = xlog_checkpoint.oldestXid;
1125 0 : }
1126 0 : trace!(
1127 0 : "xlog_checkpoint.oldestActiveXid={}, checkpoint.oldestActiveXid={}",
1128 : xlog_checkpoint.oldestActiveXid,
1129 : cp.oldestActiveXid
1130 : );
1131 :
1132 : // A shutdown checkpoint has `oldestActiveXid == InvalidTransactionid`,
1133 : // because at shutdown, all in-progress transactions will implicitly
1134 : // end. Postgres startup code knows that, and allows hot standby to start
1135 : // immediately from a shutdown checkpoint.
1136 : //
1137 : // In Neon, Postgres hot standby startup always behaves as if starting from
1138 : // an online checkpoint. It needs a valid `oldestActiveXid` value, so
1139 : // instead of overwriting self.checkpoint.oldestActiveXid with
1140 : // InvalidTransactionid from the checkpoint WAL record, update it to a
1141 : // proper value, knowing that there are no in-progress transactions at this
1142 : // point, except for prepared transactions.
1143 : //
1144 : // See also the neon code changes in the InitWalRecovery() function.
1145 0 : if xlog_checkpoint.oldestActiveXid == pg_constants::INVALID_TRANSACTION_ID
1146 0 : && info == pg_constants::XLOG_CHECKPOINT_SHUTDOWN
1147 : {
1148 0 : let oldest_active_xid = if pg_version >= 17 {
1149 0 : let mut oldest_active_full_xid = cp.nextXid.value;
1150 0 : for xid in modification.tline.list_twophase_files(lsn, ctx).await? {
1151 0 : if xid < oldest_active_full_xid {
1152 0 : oldest_active_full_xid = xid;
1153 0 : }
1154 : }
1155 0 : oldest_active_full_xid as u32
1156 : } else {
1157 0 : let mut oldest_active_xid = cp.nextXid.value as u32;
1158 0 : for xid in modification.tline.list_twophase_files(lsn, ctx).await? {
1159 0 : let narrow_xid = xid as u32;
1160 0 : if (narrow_xid.wrapping_sub(oldest_active_xid) as i32) < 0 {
1161 0 : oldest_active_xid = narrow_xid;
1162 0 : }
1163 : }
1164 0 : oldest_active_xid
1165 : };
1166 0 : cp.oldestActiveXid = oldest_active_xid;
1167 0 : } else {
1168 0 : cp.oldestActiveXid = xlog_checkpoint.oldestActiveXid;
1169 0 : }
1170 :
1171 : // Write a new checkpoint key-value pair on every checkpoint record, even
1172 : // if nothing really changed. Not strictly required, but it seems nice to
1173 : // have some trace of the checkpoint records in the layer files at the same
1174 : // LSNs.
1175 0 : self.checkpoint_modified = true;
1176 0 : }
1177 : });
1178 :
1179 30 : Ok(())
1180 30 : }
1181 :
1182 0 : async fn ingest_logical_message_put(
1183 0 : &mut self,
1184 0 : put: PutLogicalMessage,
1185 0 : modification: &mut DatadirModification<'_>,
1186 0 : ctx: &RequestContext,
1187 0 : ) -> Result<()> {
1188 0 : let PutLogicalMessage { path, buf } = put;
1189 0 : modification.put_file(path.as_str(), &buf, ctx).await
1190 0 : }
1191 :
1192 0 : fn ingest_standby_record(&mut self, record: StandbyRecord) -> Result<()> {
1193 0 : match record {
1194 0 : StandbyRecord::RunningXacts(running_xacts) => {
1195 0 : enum_pgversion_dispatch!(&mut self.checkpoint, CheckPoint, cp, {
1196 0 : cp.oldestActiveXid = running_xacts.oldest_running_xid;
1197 0 : });
1198 :
1199 0 : self.checkpoint_modified = true;
1200 0 : }
1201 0 : }
1202 0 :
1203 0 : Ok(())
1204 0 : }
1205 :
1206 0 : async fn ingest_replorigin_record(
1207 0 : &mut self,
1208 0 : record: ReploriginRecord,
1209 0 : modification: &mut DatadirModification<'_>,
1210 0 : ) -> Result<()> {
1211 0 : match record {
1212 0 : ReploriginRecord::Set(set) => {
1213 0 : modification
1214 0 : .set_replorigin(set.node_id, set.remote_lsn)
1215 0 : .await?;
1216 : }
1217 0 : ReploriginRecord::Drop(drop) => {
1218 0 : modification.drop_replorigin(drop.node_id).await?;
1219 : }
1220 : }
1221 :
1222 0 : Ok(())
1223 0 : }
1224 :
1225 18 : async fn put_rel_creation(
1226 18 : &mut self,
1227 18 : modification: &mut DatadirModification<'_>,
1228 18 : rel: RelTag,
1229 18 : ctx: &RequestContext,
1230 18 : ) -> Result<()> {
1231 18 : modification.put_rel_creation(rel, 0, ctx).await?;
1232 18 : Ok(())
1233 18 : }
1234 :
1235 : #[cfg(test)]
1236 272402 : async fn put_rel_page_image(
1237 272402 : &mut self,
1238 272402 : modification: &mut DatadirModification<'_>,
1239 272402 : rel: RelTag,
1240 272402 : blknum: BlockNumber,
1241 272402 : img: Bytes,
1242 272402 : ctx: &RequestContext,
1243 272402 : ) -> Result<(), PageReconstructError> {
1244 272402 : self.handle_rel_extend(modification, rel, blknum, ctx)
1245 5420 : .await?;
1246 272402 : modification.put_rel_page_image(rel, blknum, img)?;
1247 272402 : Ok(())
1248 272402 : }
1249 :
1250 12 : async fn put_rel_wal_record(
1251 12 : &mut self,
1252 12 : modification: &mut DatadirModification<'_>,
1253 12 : rel: RelTag,
1254 12 : blknum: BlockNumber,
1255 12 : rec: NeonWalRecord,
1256 12 : ctx: &RequestContext,
1257 12 : ) -> Result<()> {
1258 12 : self.handle_rel_extend(modification, rel, blknum, ctx)
1259 0 : .await?;
1260 12 : modification.put_rel_wal_record(rel, blknum, rec)?;
1261 12 : Ok(())
1262 12 : }
1263 :
1264 6012 : async fn put_rel_truncation(
1265 6012 : &mut self,
1266 6012 : modification: &mut DatadirModification<'_>,
1267 6012 : rel: RelTag,
1268 6012 : nblocks: BlockNumber,
1269 6012 : ctx: &RequestContext,
1270 6012 : ) -> anyhow::Result<()> {
1271 6012 : modification.put_rel_truncation(rel, nblocks, ctx).await?;
1272 6012 : Ok(())
1273 6012 : }
1274 :
1275 272414 : async fn handle_rel_extend(
1276 272414 : &mut self,
1277 272414 : modification: &mut DatadirModification<'_>,
1278 272414 : rel: RelTag,
1279 272414 : blknum: BlockNumber,
1280 272414 : ctx: &RequestContext,
1281 272414 : ) -> Result<(), PageReconstructError> {
1282 272414 : let new_nblocks = blknum + 1;
1283 : // Check if the relation exists. We implicitly create relations on first
1284 : // record.
1285 272414 : let old_nblocks = modification.create_relation_if_required(rel, ctx).await?;
1286 :
1287 272414 : if new_nblocks > old_nblocks {
1288 : //info!("extending {} {} to {}", rel, old_nblocks, new_nblocks);
1289 272398 : modification.put_rel_extend(rel, new_nblocks, ctx).await?;
1290 :
1291 272398 : let mut key = rel_block_to_key(rel, blknum);
1292 272398 :
1293 272398 : // fill the gap with zeros
1294 272398 : let mut gap_blocks_filled: u64 = 0;
1295 272398 : for gap_blknum in old_nblocks..blknum {
1296 2998 : key.field6 = gap_blknum;
1297 2998 :
1298 2998 : if self.shard.get_shard_number(&key) != self.shard.number {
1299 0 : continue;
1300 2998 : }
1301 2998 :
1302 2998 : modification.put_rel_page_image_zero(rel, gap_blknum)?;
1303 2998 : gap_blocks_filled += 1;
1304 : }
1305 :
1306 272398 : WAL_INGEST
1307 272398 : .gap_blocks_zeroed_on_rel_extend
1308 272398 : .inc_by(gap_blocks_filled);
1309 272398 :
1310 272398 : // Log something when relation extends cause use to fill gaps
1311 272398 : // with zero pages. Logging is rate limited per pg version to
1312 272398 : // avoid skewing.
1313 272398 : if gap_blocks_filled > 0 {
1314 : use once_cell::sync::Lazy;
1315 : use std::sync::Mutex;
1316 : use utils::rate_limit::RateLimit;
1317 :
1318 : struct RateLimitPerPgVersion {
1319 : rate_limiters: [Lazy<Mutex<RateLimit>>; 4],
1320 : }
1321 :
1322 : impl RateLimitPerPgVersion {
1323 0 : const fn new() -> Self {
1324 0 : Self {
1325 0 : rate_limiters: [const {
1326 2 : Lazy::new(|| Mutex::new(RateLimit::new(Duration::from_secs(30))))
1327 0 : }; 4],
1328 0 : }
1329 0 : }
1330 :
1331 4 : const fn rate_limiter(
1332 4 : &self,
1333 4 : pg_version: u32,
1334 4 : ) -> Option<&Lazy<Mutex<RateLimit>>> {
1335 : const MIN_PG_VERSION: u32 = 14;
1336 : const MAX_PG_VERSION: u32 = 17;
1337 :
1338 4 : if pg_version < MIN_PG_VERSION || pg_version > MAX_PG_VERSION {
1339 0 : return None;
1340 4 : }
1341 4 :
1342 4 : Some(&self.rate_limiters[(pg_version - MIN_PG_VERSION) as usize])
1343 4 : }
1344 : }
1345 :
1346 : static LOGGED: RateLimitPerPgVersion = RateLimitPerPgVersion::new();
1347 4 : if let Some(rate_limiter) = LOGGED.rate_limiter(modification.tline.pg_version) {
1348 4 : if let Ok(mut locked) = rate_limiter.try_lock() {
1349 4 : locked.call(|| {
1350 2 : info!(
1351 0 : lsn=%modification.get_lsn(),
1352 0 : pg_version=%modification.tline.pg_version,
1353 0 : rel=%rel,
1354 0 : "Filled {} gap blocks on rel extend to {} from {}",
1355 : gap_blocks_filled,
1356 : new_nblocks,
1357 : old_nblocks);
1358 4 : });
1359 4 : }
1360 0 : }
1361 272394 : }
1362 16 : }
1363 272414 : Ok(())
1364 272414 : }
1365 :
1366 0 : async fn put_slru_page_image(
1367 0 : &mut self,
1368 0 : modification: &mut DatadirModification<'_>,
1369 0 : kind: SlruKind,
1370 0 : segno: u32,
1371 0 : blknum: BlockNumber,
1372 0 : img: Bytes,
1373 0 : ctx: &RequestContext,
1374 0 : ) -> Result<()> {
1375 0 : self.handle_slru_extend(modification, kind, segno, blknum, ctx)
1376 0 : .await?;
1377 0 : modification.put_slru_page_image(kind, segno, blknum, img)?;
1378 0 : Ok(())
1379 0 : }
1380 :
1381 0 : async fn handle_slru_extend(
1382 0 : &mut self,
1383 0 : modification: &mut DatadirModification<'_>,
1384 0 : kind: SlruKind,
1385 0 : segno: u32,
1386 0 : blknum: BlockNumber,
1387 0 : ctx: &RequestContext,
1388 0 : ) -> anyhow::Result<()> {
1389 0 : // we don't use a cache for this like we do for relations. SLRUS are explcitly
1390 0 : // extended with ZEROPAGE records, not with commit records, so it happens
1391 0 : // a lot less frequently.
1392 0 :
1393 0 : let new_nblocks = blknum + 1;
1394 : // Check if the relation exists. We implicitly create relations on first
1395 : // record.
1396 : // TODO: would be nice if to be more explicit about it
1397 0 : let old_nblocks = if !modification
1398 0 : .tline
1399 0 : .get_slru_segment_exists(kind, segno, Version::Modified(modification), ctx)
1400 0 : .await?
1401 : {
1402 : // create it with 0 size initially, the logic below will extend it
1403 0 : modification
1404 0 : .put_slru_segment_creation(kind, segno, 0, ctx)
1405 0 : .await?;
1406 0 : 0
1407 : } else {
1408 0 : modification
1409 0 : .tline
1410 0 : .get_slru_segment_size(kind, segno, Version::Modified(modification), ctx)
1411 0 : .await?
1412 : };
1413 :
1414 0 : if new_nblocks > old_nblocks {
1415 0 : trace!(
1416 0 : "extending SLRU {:?} seg {} from {} to {} blocks",
1417 : kind,
1418 : segno,
1419 : old_nblocks,
1420 : new_nblocks
1421 : );
1422 0 : modification.put_slru_extend(kind, segno, new_nblocks)?;
1423 :
1424 : // fill the gap with zeros
1425 0 : for gap_blknum in old_nblocks..blknum {
1426 0 : modification.put_slru_page_image_zero(kind, segno, gap_blknum)?;
1427 : }
1428 0 : }
1429 0 : Ok(())
1430 0 : }
1431 : }
1432 :
1433 12 : async fn get_relsize(
1434 12 : modification: &DatadirModification<'_>,
1435 12 : rel: RelTag,
1436 12 : ctx: &RequestContext,
1437 12 : ) -> Result<BlockNumber, PageReconstructError> {
1438 12 : let nblocks = if !modification
1439 12 : .tline
1440 12 : .get_rel_exists(rel, Version::Modified(modification), ctx)
1441 0 : .await?
1442 : {
1443 0 : 0
1444 : } else {
1445 12 : modification
1446 12 : .tline
1447 12 : .get_rel_size(rel, Version::Modified(modification), ctx)
1448 0 : .await?
1449 : };
1450 12 : Ok(nblocks)
1451 12 : }
1452 :
1453 : #[allow(clippy::bool_assert_comparison)]
1454 : #[cfg(test)]
1455 : mod tests {
1456 : use super::*;
1457 : use crate::tenant::harness::*;
1458 : use crate::tenant::remote_timeline_client::{remote_initdb_archive_path, INITDB_PATH};
1459 : use postgres_ffi::RELSEG_SIZE;
1460 :
1461 : use crate::DEFAULT_PG_VERSION;
1462 :
1463 : /// Arbitrary relation tag, for testing.
1464 : const TESTREL_A: RelTag = RelTag {
1465 : spcnode: 0,
1466 : dbnode: 111,
1467 : relnode: 1000,
1468 : forknum: 0,
1469 : };
1470 :
1471 12 : fn assert_current_logical_size(_timeline: &Timeline, _lsn: Lsn) {
1472 12 : // TODO
1473 12 : }
1474 :
1475 : #[tokio::test]
1476 2 : async fn test_zeroed_checkpoint_decodes_correctly() -> Result<()> {
1477 8 : for i in 14..=16 {
1478 6 : dispatch_pgversion!(i, {
1479 2 : pgv::CheckPoint::decode(&pgv::ZERO_CHECKPOINT)?;
1480 2 : });
1481 2 : }
1482 2 :
1483 2 : Ok(())
1484 2 : }
1485 :
1486 8 : async fn init_walingest_test(tline: &Timeline, ctx: &RequestContext) -> Result<WalIngest> {
1487 8 : let mut m = tline.begin_modification(Lsn(0x10));
1488 8 : m.put_checkpoint(dispatch_pgversion!(
1489 8 : tline.pg_version,
1490 0 : pgv::ZERO_CHECKPOINT.clone()
1491 0 : ))?;
1492 16 : m.put_relmap_file(0, 111, Bytes::from(""), ctx).await?; // dummy relmapper file
1493 8 : m.commit(ctx).await?;
1494 8 : let walingest = WalIngest::new(tline, Lsn(0x10), ctx).await?;
1495 :
1496 8 : Ok(walingest)
1497 8 : }
1498 :
1499 : #[tokio::test]
1500 2 : async fn test_relsize() -> Result<()> {
1501 20 : let (tenant, ctx) = TenantHarness::create("test_relsize").await?.load().await;
1502 2 : let tline = tenant
1503 2 : .create_test_timeline(TIMELINE_ID, Lsn(8), DEFAULT_PG_VERSION, &ctx)
1504 6 : .await?;
1505 5 : let mut walingest = init_walingest_test(&tline, &ctx).await?;
1506 2 :
1507 2 : let mut m = tline.begin_modification(Lsn(0x20));
1508 2 : walingest.put_rel_creation(&mut m, TESTREL_A, &ctx).await?;
1509 2 : walingest
1510 2 : .put_rel_page_image(&mut m, TESTREL_A, 0, test_img("foo blk 0 at 2"), &ctx)
1511 2 : .await?;
1512 2 : m.commit(&ctx).await?;
1513 2 : let mut m = tline.begin_modification(Lsn(0x30));
1514 2 : walingest
1515 2 : .put_rel_page_image(&mut m, TESTREL_A, 0, test_img("foo blk 0 at 3"), &ctx)
1516 2 : .await?;
1517 2 : m.commit(&ctx).await?;
1518 2 : let mut m = tline.begin_modification(Lsn(0x40));
1519 2 : walingest
1520 2 : .put_rel_page_image(&mut m, TESTREL_A, 1, test_img("foo blk 1 at 4"), &ctx)
1521 2 : .await?;
1522 2 : m.commit(&ctx).await?;
1523 2 : let mut m = tline.begin_modification(Lsn(0x50));
1524 2 : walingest
1525 2 : .put_rel_page_image(&mut m, TESTREL_A, 2, test_img("foo blk 2 at 5"), &ctx)
1526 2 : .await?;
1527 2 : m.commit(&ctx).await?;
1528 2 :
1529 2 : assert_current_logical_size(&tline, Lsn(0x50));
1530 2 :
1531 2 : let test_span = tracing::info_span!(parent: None, "test",
1532 2 : tenant_id=%tline.tenant_shard_id.tenant_id,
1533 0 : shard_id=%tline.tenant_shard_id.shard_slug(),
1534 0 : timeline_id=%tline.timeline_id);
1535 2 :
1536 2 : // The relation was created at LSN 2, not visible at LSN 1 yet.
1537 2 : assert_eq!(
1538 2 : tline
1539 2 : .get_rel_exists(TESTREL_A, Version::Lsn(Lsn(0x10)), &ctx)
1540 2 : .await?,
1541 2 : false
1542 2 : );
1543 2 : assert!(tline
1544 2 : .get_rel_size(TESTREL_A, Version::Lsn(Lsn(0x10)), &ctx)
1545 2 : .await
1546 2 : .is_err());
1547 2 : assert_eq!(
1548 2 : tline
1549 2 : .get_rel_exists(TESTREL_A, Version::Lsn(Lsn(0x20)), &ctx)
1550 2 : .await?,
1551 2 : true
1552 2 : );
1553 2 : assert_eq!(
1554 2 : tline
1555 2 : .get_rel_size(TESTREL_A, Version::Lsn(Lsn(0x20)), &ctx)
1556 2 : .await?,
1557 2 : 1
1558 2 : );
1559 2 : assert_eq!(
1560 2 : tline
1561 2 : .get_rel_size(TESTREL_A, Version::Lsn(Lsn(0x50)), &ctx)
1562 2 : .await?,
1563 2 : 3
1564 2 : );
1565 2 :
1566 2 : // Check page contents at each LSN
1567 2 : assert_eq!(
1568 2 : tline
1569 2 : .get_rel_page_at_lsn(TESTREL_A, 0, Version::Lsn(Lsn(0x20)), &ctx)
1570 2 : .instrument(test_span.clone())
1571 2 : .await?,
1572 2 : test_img("foo blk 0 at 2")
1573 2 : );
1574 2 :
1575 2 : assert_eq!(
1576 2 : tline
1577 2 : .get_rel_page_at_lsn(TESTREL_A, 0, Version::Lsn(Lsn(0x30)), &ctx)
1578 2 : .instrument(test_span.clone())
1579 2 : .await?,
1580 2 : test_img("foo blk 0 at 3")
1581 2 : );
1582 2 :
1583 2 : assert_eq!(
1584 2 : tline
1585 2 : .get_rel_page_at_lsn(TESTREL_A, 0, Version::Lsn(Lsn(0x40)), &ctx)
1586 2 : .instrument(test_span.clone())
1587 2 : .await?,
1588 2 : test_img("foo blk 0 at 3")
1589 2 : );
1590 2 : assert_eq!(
1591 2 : tline
1592 2 : .get_rel_page_at_lsn(TESTREL_A, 1, Version::Lsn(Lsn(0x40)), &ctx)
1593 2 : .instrument(test_span.clone())
1594 2 : .await?,
1595 2 : test_img("foo blk 1 at 4")
1596 2 : );
1597 2 :
1598 2 : assert_eq!(
1599 2 : tline
1600 2 : .get_rel_page_at_lsn(TESTREL_A, 0, Version::Lsn(Lsn(0x50)), &ctx)
1601 2 : .instrument(test_span.clone())
1602 2 : .await?,
1603 2 : test_img("foo blk 0 at 3")
1604 2 : );
1605 2 : assert_eq!(
1606 2 : tline
1607 2 : .get_rel_page_at_lsn(TESTREL_A, 1, Version::Lsn(Lsn(0x50)), &ctx)
1608 2 : .instrument(test_span.clone())
1609 2 : .await?,
1610 2 : test_img("foo blk 1 at 4")
1611 2 : );
1612 2 : assert_eq!(
1613 2 : tline
1614 2 : .get_rel_page_at_lsn(TESTREL_A, 2, Version::Lsn(Lsn(0x50)), &ctx)
1615 2 : .instrument(test_span.clone())
1616 2 : .await?,
1617 2 : test_img("foo blk 2 at 5")
1618 2 : );
1619 2 :
1620 2 : // Truncate last block
1621 2 : let mut m = tline.begin_modification(Lsn(0x60));
1622 2 : walingest
1623 2 : .put_rel_truncation(&mut m, TESTREL_A, 2, &ctx)
1624 2 : .await?;
1625 2 : m.commit(&ctx).await?;
1626 2 : assert_current_logical_size(&tline, Lsn(0x60));
1627 2 :
1628 2 : // Check reported size and contents after truncation
1629 2 : assert_eq!(
1630 2 : tline
1631 2 : .get_rel_size(TESTREL_A, Version::Lsn(Lsn(0x60)), &ctx)
1632 2 : .await?,
1633 2 : 2
1634 2 : );
1635 2 : assert_eq!(
1636 2 : tline
1637 2 : .get_rel_page_at_lsn(TESTREL_A, 0, Version::Lsn(Lsn(0x60)), &ctx)
1638 2 : .instrument(test_span.clone())
1639 2 : .await?,
1640 2 : test_img("foo blk 0 at 3")
1641 2 : );
1642 2 : assert_eq!(
1643 2 : tline
1644 2 : .get_rel_page_at_lsn(TESTREL_A, 1, Version::Lsn(Lsn(0x60)), &ctx)
1645 2 : .instrument(test_span.clone())
1646 2 : .await?,
1647 2 : test_img("foo blk 1 at 4")
1648 2 : );
1649 2 :
1650 2 : // should still see the truncated block with older LSN
1651 2 : assert_eq!(
1652 2 : tline
1653 2 : .get_rel_size(TESTREL_A, Version::Lsn(Lsn(0x50)), &ctx)
1654 2 : .await?,
1655 2 : 3
1656 2 : );
1657 2 : assert_eq!(
1658 2 : tline
1659 2 : .get_rel_page_at_lsn(TESTREL_A, 2, Version::Lsn(Lsn(0x50)), &ctx)
1660 2 : .instrument(test_span.clone())
1661 2 : .await?,
1662 2 : test_img("foo blk 2 at 5")
1663 2 : );
1664 2 :
1665 2 : // Truncate to zero length
1666 2 : let mut m = tline.begin_modification(Lsn(0x68));
1667 2 : walingest
1668 2 : .put_rel_truncation(&mut m, TESTREL_A, 0, &ctx)
1669 2 : .await?;
1670 2 : m.commit(&ctx).await?;
1671 2 : assert_eq!(
1672 2 : tline
1673 2 : .get_rel_size(TESTREL_A, Version::Lsn(Lsn(0x68)), &ctx)
1674 2 : .await?,
1675 2 : 0
1676 2 : );
1677 2 :
1678 2 : // Extend from 0 to 2 blocks, leaving a gap
1679 2 : let mut m = tline.begin_modification(Lsn(0x70));
1680 2 : walingest
1681 2 : .put_rel_page_image(&mut m, TESTREL_A, 1, test_img("foo blk 1"), &ctx)
1682 2 : .await?;
1683 2 : m.commit(&ctx).await?;
1684 2 : assert_eq!(
1685 2 : tline
1686 2 : .get_rel_size(TESTREL_A, Version::Lsn(Lsn(0x70)), &ctx)
1687 2 : .await?,
1688 2 : 2
1689 2 : );
1690 2 : assert_eq!(
1691 2 : tline
1692 2 : .get_rel_page_at_lsn(TESTREL_A, 0, Version::Lsn(Lsn(0x70)), &ctx)
1693 2 : .instrument(test_span.clone())
1694 2 : .await?,
1695 2 : ZERO_PAGE
1696 2 : );
1697 2 : assert_eq!(
1698 2 : tline
1699 2 : .get_rel_page_at_lsn(TESTREL_A, 1, Version::Lsn(Lsn(0x70)), &ctx)
1700 2 : .instrument(test_span.clone())
1701 2 : .await?,
1702 2 : test_img("foo blk 1")
1703 2 : );
1704 2 :
1705 2 : // Extend a lot more, leaving a big gap that spans across segments
1706 2 : let mut m = tline.begin_modification(Lsn(0x80));
1707 2 : walingest
1708 2 : .put_rel_page_image(&mut m, TESTREL_A, 1500, test_img("foo blk 1500"), &ctx)
1709 2 : .await?;
1710 190 : m.commit(&ctx).await?;
1711 2 : assert_eq!(
1712 2 : tline
1713 2 : .get_rel_size(TESTREL_A, Version::Lsn(Lsn(0x80)), &ctx)
1714 2 : .await?,
1715 2 : 1501
1716 2 : );
1717 2998 : for blk in 2..1500 {
1718 2996 : assert_eq!(
1719 2996 : tline
1720 2996 : .get_rel_page_at_lsn(TESTREL_A, blk, Version::Lsn(Lsn(0x80)), &ctx)
1721 2996 : .instrument(test_span.clone())
1722 1540 : .await?,
1723 2996 : ZERO_PAGE
1724 2 : );
1725 2 : }
1726 2 : assert_eq!(
1727 2 : tline
1728 2 : .get_rel_page_at_lsn(TESTREL_A, 1500, Version::Lsn(Lsn(0x80)), &ctx)
1729 2 : .instrument(test_span.clone())
1730 2 : .await?,
1731 2 : test_img("foo blk 1500")
1732 2 : );
1733 2 :
1734 2 : Ok(())
1735 2 : }
1736 :
1737 : // Test what happens if we dropped a relation
1738 : // and then created it again within the same layer.
1739 : #[tokio::test]
1740 2 : async fn test_drop_extend() -> Result<()> {
1741 2 : let (tenant, ctx) = TenantHarness::create("test_drop_extend")
1742 2 : .await?
1743 2 : .load()
1744 20 : .await;
1745 2 : let tline = tenant
1746 2 : .create_test_timeline(TIMELINE_ID, Lsn(8), DEFAULT_PG_VERSION, &ctx)
1747 6 : .await?;
1748 5 : let mut walingest = init_walingest_test(&tline, &ctx).await?;
1749 2 :
1750 2 : let mut m = tline.begin_modification(Lsn(0x20));
1751 2 : walingest
1752 2 : .put_rel_page_image(&mut m, TESTREL_A, 0, test_img("foo blk 0 at 2"), &ctx)
1753 2 : .await?;
1754 2 : m.commit(&ctx).await?;
1755 2 :
1756 2 : // Check that rel exists and size is correct
1757 2 : assert_eq!(
1758 2 : tline
1759 2 : .get_rel_exists(TESTREL_A, Version::Lsn(Lsn(0x20)), &ctx)
1760 2 : .await?,
1761 2 : true
1762 2 : );
1763 2 : assert_eq!(
1764 2 : tline
1765 2 : .get_rel_size(TESTREL_A, Version::Lsn(Lsn(0x20)), &ctx)
1766 2 : .await?,
1767 2 : 1
1768 2 : );
1769 2 :
1770 2 : // Drop rel
1771 2 : let mut m = tline.begin_modification(Lsn(0x30));
1772 2 : let mut rel_drops = HashMap::new();
1773 2 : rel_drops.insert((TESTREL_A.spcnode, TESTREL_A.dbnode), vec![TESTREL_A]);
1774 2 : m.put_rel_drops(rel_drops, &ctx).await?;
1775 2 : m.commit(&ctx).await?;
1776 2 :
1777 2 : // Check that rel is not visible anymore
1778 2 : assert_eq!(
1779 2 : tline
1780 2 : .get_rel_exists(TESTREL_A, Version::Lsn(Lsn(0x30)), &ctx)
1781 2 : .await?,
1782 2 : false
1783 2 : );
1784 2 :
1785 2 : // FIXME: should fail
1786 2 : //assert!(tline.get_rel_size(TESTREL_A, Lsn(0x30), false)?.is_none());
1787 2 :
1788 2 : // Re-create it
1789 2 : let mut m = tline.begin_modification(Lsn(0x40));
1790 2 : walingest
1791 2 : .put_rel_page_image(&mut m, TESTREL_A, 0, test_img("foo blk 0 at 4"), &ctx)
1792 2 : .await?;
1793 2 : m.commit(&ctx).await?;
1794 2 :
1795 2 : // Check that rel exists and size is correct
1796 2 : assert_eq!(
1797 2 : tline
1798 2 : .get_rel_exists(TESTREL_A, Version::Lsn(Lsn(0x40)), &ctx)
1799 2 : .await?,
1800 2 : true
1801 2 : );
1802 2 : assert_eq!(
1803 2 : tline
1804 2 : .get_rel_size(TESTREL_A, Version::Lsn(Lsn(0x40)), &ctx)
1805 2 : .await?,
1806 2 : 1
1807 2 : );
1808 2 :
1809 2 : Ok(())
1810 2 : }
1811 :
1812 : // Test what happens if we truncated a relation
1813 : // so that one of its segments was dropped
1814 : // and then extended it again within the same layer.
1815 : #[tokio::test]
1816 2 : async fn test_truncate_extend() -> Result<()> {
1817 2 : let (tenant, ctx) = TenantHarness::create("test_truncate_extend")
1818 2 : .await?
1819 2 : .load()
1820 20 : .await;
1821 2 : let tline = tenant
1822 2 : .create_test_timeline(TIMELINE_ID, Lsn(8), DEFAULT_PG_VERSION, &ctx)
1823 6 : .await?;
1824 5 : let mut walingest = init_walingest_test(&tline, &ctx).await?;
1825 2 :
1826 2 : // Create a 20 MB relation (the size is arbitrary)
1827 2 : let relsize = 20 * 1024 * 1024 / 8192;
1828 2 : let mut m = tline.begin_modification(Lsn(0x20));
1829 5120 : for blkno in 0..relsize {
1830 5120 : let data = format!("foo blk {} at {}", blkno, Lsn(0x20));
1831 5120 : walingest
1832 5120 : .put_rel_page_image(&mut m, TESTREL_A, blkno, test_img(&data), &ctx)
1833 2 : .await?;
1834 2 : }
1835 2 : m.commit(&ctx).await?;
1836 2 :
1837 2 : let test_span = tracing::info_span!(parent: None, "test",
1838 2 : tenant_id=%tline.tenant_shard_id.tenant_id,
1839 0 : shard_id=%tline.tenant_shard_id.shard_slug(),
1840 0 : timeline_id=%tline.timeline_id);
1841 2 :
1842 2 : // The relation was created at LSN 20, not visible at LSN 1 yet.
1843 2 : assert_eq!(
1844 2 : tline
1845 2 : .get_rel_exists(TESTREL_A, Version::Lsn(Lsn(0x10)), &ctx)
1846 2 : .await?,
1847 2 : false
1848 2 : );
1849 2 : assert!(tline
1850 2 : .get_rel_size(TESTREL_A, Version::Lsn(Lsn(0x10)), &ctx)
1851 2 : .await
1852 2 : .is_err());
1853 2 :
1854 2 : assert_eq!(
1855 2 : tline
1856 2 : .get_rel_exists(TESTREL_A, Version::Lsn(Lsn(0x20)), &ctx)
1857 2 : .await?,
1858 2 : true
1859 2 : );
1860 2 : assert_eq!(
1861 2 : tline
1862 2 : .get_rel_size(TESTREL_A, Version::Lsn(Lsn(0x20)), &ctx)
1863 2 : .await?,
1864 2 : relsize
1865 2 : );
1866 2 :
1867 2 : // Check relation content
1868 5120 : for blkno in 0..relsize {
1869 5120 : let lsn = Lsn(0x20);
1870 5120 : let data = format!("foo blk {} at {}", blkno, lsn);
1871 5120 : assert_eq!(
1872 5120 : tline
1873 5120 : .get_rel_page_at_lsn(TESTREL_A, blkno, Version::Lsn(lsn), &ctx)
1874 5120 : .instrument(test_span.clone())
1875 1803 : .await?,
1876 5120 : test_img(&data)
1877 2 : );
1878 2 : }
1879 2 :
1880 2 : // Truncate relation so that second segment was dropped
1881 2 : // - only leave one page
1882 2 : let mut m = tline.begin_modification(Lsn(0x60));
1883 2 : walingest
1884 2 : .put_rel_truncation(&mut m, TESTREL_A, 1, &ctx)
1885 2 : .await?;
1886 2 : m.commit(&ctx).await?;
1887 2 :
1888 2 : // Check reported size and contents after truncation
1889 2 : assert_eq!(
1890 2 : tline
1891 2 : .get_rel_size(TESTREL_A, Version::Lsn(Lsn(0x60)), &ctx)
1892 2 : .await?,
1893 2 : 1
1894 2 : );
1895 2 :
1896 4 : for blkno in 0..1 {
1897 2 : let lsn = Lsn(0x20);
1898 2 : let data = format!("foo blk {} at {}", blkno, lsn);
1899 2 : assert_eq!(
1900 2 : tline
1901 2 : .get_rel_page_at_lsn(TESTREL_A, blkno, Version::Lsn(Lsn(0x60)), &ctx)
1902 2 : .instrument(test_span.clone())
1903 2 : .await?,
1904 2 : test_img(&data)
1905 2 : );
1906 2 : }
1907 2 :
1908 2 : // should still see all blocks with older LSN
1909 2 : assert_eq!(
1910 2 : tline
1911 2 : .get_rel_size(TESTREL_A, Version::Lsn(Lsn(0x50)), &ctx)
1912 2 : .await?,
1913 2 : relsize
1914 2 : );
1915 5120 : for blkno in 0..relsize {
1916 5120 : let lsn = Lsn(0x20);
1917 5120 : let data = format!("foo blk {} at {}", blkno, lsn);
1918 5120 : assert_eq!(
1919 5120 : tline
1920 5120 : .get_rel_page_at_lsn(TESTREL_A, blkno, Version::Lsn(Lsn(0x50)), &ctx)
1921 5120 : .instrument(test_span.clone())
1922 1856 : .await?,
1923 5120 : test_img(&data)
1924 2 : );
1925 2 : }
1926 2 :
1927 2 : // Extend relation again.
1928 2 : // Add enough blocks to create second segment
1929 2 : let lsn = Lsn(0x80);
1930 2 : let mut m = tline.begin_modification(lsn);
1931 5120 : for blkno in 0..relsize {
1932 5120 : let data = format!("foo blk {} at {}", blkno, lsn);
1933 5120 : walingest
1934 5120 : .put_rel_page_image(&mut m, TESTREL_A, blkno, test_img(&data), &ctx)
1935 2 : .await?;
1936 2 : }
1937 3 : m.commit(&ctx).await?;
1938 2 :
1939 2 : assert_eq!(
1940 2 : tline
1941 2 : .get_rel_exists(TESTREL_A, Version::Lsn(Lsn(0x80)), &ctx)
1942 2 : .await?,
1943 2 : true
1944 2 : );
1945 2 : assert_eq!(
1946 2 : tline
1947 2 : .get_rel_size(TESTREL_A, Version::Lsn(Lsn(0x80)), &ctx)
1948 2 : .await?,
1949 2 : relsize
1950 2 : );
1951 2 : // Check relation content
1952 5120 : for blkno in 0..relsize {
1953 5120 : let lsn = Lsn(0x80);
1954 5120 : let data = format!("foo blk {} at {}", blkno, lsn);
1955 5120 : assert_eq!(
1956 5120 : tline
1957 5120 : .get_rel_page_at_lsn(TESTREL_A, blkno, Version::Lsn(Lsn(0x80)), &ctx)
1958 5120 : .instrument(test_span.clone())
1959 1828 : .await?,
1960 5120 : test_img(&data)
1961 2 : );
1962 2 : }
1963 2 :
1964 2 : Ok(())
1965 2 : }
1966 :
1967 : /// Test get_relsize() and truncation with a file larger than 1 GB, so that it's
1968 : /// split into multiple 1 GB segments in Postgres.
1969 : #[tokio::test]
1970 2 : async fn test_large_rel() -> Result<()> {
1971 20 : let (tenant, ctx) = TenantHarness::create("test_large_rel").await?.load().await;
1972 2 : let tline = tenant
1973 2 : .create_test_timeline(TIMELINE_ID, Lsn(8), DEFAULT_PG_VERSION, &ctx)
1974 6 : .await?;
1975 5 : let mut walingest = init_walingest_test(&tline, &ctx).await?;
1976 2 :
1977 2 : let mut lsn = 0x10;
1978 262146 : for blknum in 0..RELSEG_SIZE + 1 {
1979 262146 : lsn += 0x10;
1980 262146 : let mut m = tline.begin_modification(Lsn(lsn));
1981 262146 : let img = test_img(&format!("foo blk {} at {}", blknum, Lsn(lsn)));
1982 262146 : walingest
1983 262146 : .put_rel_page_image(&mut m, TESTREL_A, blknum as BlockNumber, img, &ctx)
1984 5420 : .await?;
1985 262146 : m.commit(&ctx).await?;
1986 2 : }
1987 2 :
1988 2 : assert_current_logical_size(&tline, Lsn(lsn));
1989 2 :
1990 2 : assert_eq!(
1991 2 : tline
1992 2 : .get_rel_size(TESTREL_A, Version::Lsn(Lsn(lsn)), &ctx)
1993 2 : .await?,
1994 2 : RELSEG_SIZE + 1
1995 2 : );
1996 2 :
1997 2 : // Truncate one block
1998 2 : lsn += 0x10;
1999 2 : let mut m = tline.begin_modification(Lsn(lsn));
2000 2 : walingest
2001 2 : .put_rel_truncation(&mut m, TESTREL_A, RELSEG_SIZE, &ctx)
2002 2 : .await?;
2003 2 : m.commit(&ctx).await?;
2004 2 : assert_eq!(
2005 2 : tline
2006 2 : .get_rel_size(TESTREL_A, Version::Lsn(Lsn(lsn)), &ctx)
2007 2 : .await?,
2008 2 : RELSEG_SIZE
2009 2 : );
2010 2 : assert_current_logical_size(&tline, Lsn(lsn));
2011 2 :
2012 2 : // Truncate another block
2013 2 : lsn += 0x10;
2014 2 : let mut m = tline.begin_modification(Lsn(lsn));
2015 2 : walingest
2016 2 : .put_rel_truncation(&mut m, TESTREL_A, RELSEG_SIZE - 1, &ctx)
2017 2 : .await?;
2018 2 : m.commit(&ctx).await?;
2019 2 : assert_eq!(
2020 2 : tline
2021 2 : .get_rel_size(TESTREL_A, Version::Lsn(Lsn(lsn)), &ctx)
2022 2 : .await?,
2023 2 : RELSEG_SIZE - 1
2024 2 : );
2025 2 : assert_current_logical_size(&tline, Lsn(lsn));
2026 2 :
2027 2 : // Truncate to 1500, and then truncate all the way down to 0, one block at a time
2028 2 : // This tests the behavior at segment boundaries
2029 2 : let mut size: i32 = 3000;
2030 6004 : while size >= 0 {
2031 6002 : lsn += 0x10;
2032 6002 : let mut m = tline.begin_modification(Lsn(lsn));
2033 6002 : walingest
2034 6002 : .put_rel_truncation(&mut m, TESTREL_A, size as BlockNumber, &ctx)
2035 2 : .await?;
2036 6002 : m.commit(&ctx).await?;
2037 6002 : assert_eq!(
2038 6002 : tline
2039 6002 : .get_rel_size(TESTREL_A, Version::Lsn(Lsn(lsn)), &ctx)
2040 2 : .await?,
2041 6002 : size as BlockNumber
2042 2 : );
2043 2 :
2044 6002 : size -= 1;
2045 2 : }
2046 2 : assert_current_logical_size(&tline, Lsn(lsn));
2047 2 :
2048 2 : Ok(())
2049 2 : }
2050 :
2051 : /// Replay a wal segment file taken directly from safekeepers.
2052 : ///
2053 : /// This test is useful for benchmarking since it allows us to profile only
2054 : /// the walingest code in a single-threaded executor, and iterate more quickly
2055 : /// without waiting for unrelated steps.
2056 : #[tokio::test]
2057 2 : async fn test_ingest_real_wal() {
2058 2 : use crate::tenant::harness::*;
2059 2 : use postgres_ffi::waldecoder::WalStreamDecoder;
2060 2 : use postgres_ffi::WAL_SEGMENT_SIZE;
2061 2 :
2062 2 : // Define test data path and constants.
2063 2 : //
2064 2 : // Steps to reconstruct the data, if needed:
2065 2 : // 1. Run the pgbench python test
2066 2 : // 2. Take the first wal segment file from safekeeper
2067 2 : // 3. Compress it using `zstd --long input_file`
2068 2 : // 4. Copy initdb.tar.zst from local_fs_remote_storage
2069 2 : // 5. Grep sk logs for "restart decoder" to get startpoint
2070 2 : // 6. Run just the decoder from this test to get the endpoint.
2071 2 : // It's the last LSN the decoder will output.
2072 2 : let pg_version = 15; // The test data was generated by pg15
2073 2 : let path = "test_data/sk_wal_segment_from_pgbench";
2074 2 : let wal_segment_path = format!("{path}/000000010000000000000001.zst");
2075 2 : let source_initdb_path = format!("{path}/{INITDB_PATH}");
2076 2 : let startpoint = Lsn::from_hex("14AEC08").unwrap();
2077 2 : let _endpoint = Lsn::from_hex("1FFFF98").unwrap();
2078 2 :
2079 2 : let harness = TenantHarness::create("test_ingest_real_wal").await.unwrap();
2080 2 : let span = harness
2081 2 : .span()
2082 2 : .in_scope(|| info_span!("timeline_span", timeline_id=%TIMELINE_ID));
2083 20 : let (tenant, ctx) = harness.load().await;
2084 2 :
2085 2 : let remote_initdb_path =
2086 2 : remote_initdb_archive_path(&tenant.tenant_shard_id().tenant_id, &TIMELINE_ID);
2087 2 : let initdb_path = harness.remote_fs_dir.join(remote_initdb_path.get_path());
2088 2 :
2089 2 : std::fs::create_dir_all(initdb_path.parent().unwrap())
2090 2 : .expect("creating test dir should work");
2091 2 : std::fs::copy(source_initdb_path, initdb_path).expect("copying the initdb.tar.zst works");
2092 2 :
2093 2 : // Bootstrap a real timeline. We can't use create_test_timeline because
2094 2 : // it doesn't create a real checkpoint, and Walingest::new tries to parse
2095 2 : // the garbage data.
2096 2 : let tline = tenant
2097 2 : .bootstrap_timeline_test(TIMELINE_ID, pg_version, Some(TIMELINE_ID), &ctx)
2098 20103 : .await
2099 2 : .unwrap();
2100 2 :
2101 2 : // We fully read and decompress this into memory before decoding
2102 2 : // to get a more accurate perf profile of the decoder.
2103 2 : let bytes = {
2104 2 : use async_compression::tokio::bufread::ZstdDecoder;
2105 2 : let file = tokio::fs::File::open(wal_segment_path).await.unwrap();
2106 2 : let reader = tokio::io::BufReader::new(file);
2107 2 : let decoder = ZstdDecoder::new(reader);
2108 2 : let mut reader = tokio::io::BufReader::new(decoder);
2109 2 : let mut buffer = Vec::new();
2110 224 : tokio::io::copy_buf(&mut reader, &mut buffer).await.unwrap();
2111 2 : buffer
2112 2 : };
2113 2 :
2114 2 : // TODO start a profiler too
2115 2 : let started_at = std::time::Instant::now();
2116 2 :
2117 2 : // Initialize walingest
2118 2 : let xlogoff: usize = startpoint.segment_offset(WAL_SEGMENT_SIZE);
2119 2 : let mut decoder = WalStreamDecoder::new(startpoint, pg_version);
2120 2 : let mut walingest = WalIngest::new(tline.as_ref(), startpoint, &ctx)
2121 5 : .await
2122 2 : .unwrap();
2123 2 : let mut modification = tline.begin_modification(startpoint);
2124 2 : println!("decoding {} bytes", bytes.len() - xlogoff);
2125 2 :
2126 2 : // Decode and ingest wal. We process the wal in chunks because
2127 2 : // that's what happens when we get bytes from safekeepers.
2128 474686 : for chunk in bytes[xlogoff..].chunks(50) {
2129 474686 : decoder.feed_bytes(chunk);
2130 620536 : while let Some((lsn, recdata)) = decoder.poll_decode().unwrap() {
2131 145850 : let interpreted = InterpretedWalRecord::from_bytes_filtered(
2132 145850 : recdata,
2133 145850 : modification.tline.get_shard_identity(),
2134 145850 : lsn,
2135 145850 : modification.tline.pg_version,
2136 145850 : )
2137 145850 : .unwrap();
2138 145850 :
2139 145850 : walingest
2140 145850 : .ingest_record(interpreted, &mut modification, &ctx)
2141 145850 : .instrument(span.clone())
2142 296 : .await
2143 145850 : .unwrap();
2144 2 : }
2145 474686 : modification.commit(&ctx).await.unwrap();
2146 2 : }
2147 2 :
2148 2 : let duration = started_at.elapsed();
2149 2 : println!("done in {:?}", duration);
2150 2 : }
2151 : }
|
