This commit is contained in:
Heikki Linnakangas
2022-03-09 11:36:39 +02:00
parent 798ff26fb0
commit 356f716d39
5 changed files with 19 additions and 13 deletions
+2 -2
View File
@@ -242,9 +242,9 @@ fn bootstrap_timeline<R: Repository>(
// Because we know it upfront avoid having an option or dummy zero value by passing it to create_empty_timeline.
let timeline = repo.create_empty_timeline(tli, lsn)?;
let page_tline: DatadirTimeline<R> = DatadirTimeline::new(timeline);
let mut page_tline: DatadirTimeline<R> = DatadirTimeline::new(timeline);
import_datadir::import_timeline_from_postgres_datadir(&pgdata_path, &page_tline, lsn)?;
import_datadir::import_timeline_from_postgres_datadir(&pgdata_path, &mut page_tline, lsn)?;
page_tline.tline.checkpoint(CheckpointConfig::Forced)?;
println!(
+2 -2
View File
@@ -30,7 +30,7 @@ use zenith_utils::lsn::Lsn;
/// cluster was not shut down cleanly.
pub fn import_timeline_from_postgres_datadir<R: Repository>(
path: &Path,
tline: &DatadirTimeline<R>,
tline: &mut DatadirTimeline<R>,
lsn: Lsn,
) -> Result<()> {
let mut pg_control: Option<ControlFileData> = None;
@@ -335,7 +335,7 @@ fn import_slru_file<R: Repository>(
/// 'startpoint' and 'endpoint' into the repository.
fn import_wal<R: Repository>(
walpath: &Path,
tline: &DatadirTimeline<R>,
tline: &mut DatadirTimeline<R>,
startpoint: Lsn,
endpoint: Lsn,
) -> Result<()> {
+9 -5
View File
@@ -1028,10 +1028,11 @@ impl LayeredTimeline {
if prev_lsn <= cont_lsn {
// Didn't make any progress in last iteration. Error out to avoid
// getting stuck in the loop.
bail!("could not find layer with more data for key {} at LSN {}, request LSN {}",
bail!("could not find layer with more data for key {} at LSN {}, request LSN {}, ancestor {}",
key,
Lsn(cont_lsn.0 - 1),
request_lsn)
request_lsn,
timeline.ancestor_lsn)
}
prev_lsn = cont_lsn;
}
@@ -1047,7 +1048,7 @@ impl LayeredTimeline {
// Recurse into ancestor if needed
if Lsn(cont_lsn.0 - 1) <= timeline.ancestor_lsn {
//info!("going into ancestor {}", timeline.ancestor_lsn);
info!("going into ancestor {}, cont_lsn is {}", timeline.ancestor_lsn, cont_lsn);
let ancestor = timeline.get_ancestor_timeline()?;
timeline_owned = ancestor;
timeline = &*timeline_owned;
@@ -1060,7 +1061,7 @@ impl LayeredTimeline {
// Check the open and frozen in-memory layers first
if let Some(open_layer) = &layers.open_layer {
let start_lsn = open_layer.get_lsn_range().start;
if cont_lsn >= start_lsn {
if cont_lsn > start_lsn {
//info!("CHECKING for {} at {} on open layer {}", reconstruct_state.key, reconstruct_state.lsn, open_layer.filename().display());
result = open_layer.get_value_reconstruct_data(key, open_layer.get_lsn_range().start..cont_lsn, reconstruct_state)?;
cont_lsn = open_layer.get_lsn_range().start;
@@ -1069,7 +1070,7 @@ impl LayeredTimeline {
}
if let Some(frozen_layer) = &layers.frozen_layer {
let start_lsn = frozen_layer.get_lsn_range().start;
if cont_lsn >= start_lsn {
if cont_lsn > start_lsn {
//info!("CHECKING for {} at {} on frozen layer {}", reconstruct_state.key, reconstruct_state.lsn, frozen_layer.filename().display());
result = frozen_layer.get_value_reconstruct_data(key, frozen_layer.get_lsn_range().start..cont_lsn, reconstruct_state)?;
cont_lsn = frozen_layer.get_lsn_range().start;
@@ -1594,6 +1595,9 @@ impl LayeredTimeline {
fn gc(&self) -> Result<GcResult> {
let now = Instant::now();
let mut result: GcResult = Default::default();
if true {
return Ok(result);
}
let disk_consistent_lsn = self.get_disk_consistent_lsn();
let _checkpoint_cs = self.checkpoint_cs.lock().unwrap();
+5 -3
View File
@@ -22,6 +22,7 @@ use std::sync::{Arc, RwLockReadGuard};
use tracing::{debug, info, warn};
use zenith_utils::bin_ser::BeSer;
use zenith_utils::lsn::{Lsn, RecordLsn};
use zenith_utils::lsn::AtomicLsn;
/// Block number within a relation or SRU. This matches PostgreSQL's BlockNumber type.
pub type BlockNumber = u32;
@@ -31,7 +32,7 @@ where
R: Repository,
{
pub tline: Arc<R::Timeline>,
pub last_partitioning: Option<Lsn>,
pub last_partitioning: AtomicLsn,
}
#[derive(Debug, Serialize, Deserialize)]
@@ -71,7 +72,7 @@ impl<R: Repository> DatadirTimeline<R> {
pub fn new(tline: Arc<R::Timeline>) -> Self {
DatadirTimeline {
tline,
last_partitioning: None,
last_partitioning: AtomicLsn::new(0),
}
}
@@ -692,7 +693,7 @@ impl<'a, R: Repository> DatadirTimelineWriter<'a, R> {
pub fn finish(self) -> Result<()> {
let writer = self.tline.tline.writer();
let last_partitioning = self.last_partitioning.unwrap_or(Lsn(0));
let last_partitioning = self.last_partitioning.load();
for (key, value) in self.pending_updates {
writer.put(key, self.lsn, value)?;
@@ -707,6 +708,7 @@ impl<'a, R: Repository> DatadirTimelineWriter<'a, R> {
let mut partitioning = self.tline.collect_keyspace(self.lsn)?;
partitioning.repartition(TARGET_FILE_SIZE_BYTES);
self.tline.tline.hint_partitioning(partitioning)?;
self.tline.last_partitioning.store(self.lsn);
}
Ok(())
+1 -1
View File
@@ -242,7 +242,7 @@ fn walreceiver_main(
let startlsn = Lsn::from(xlog_data.wal_start());
let endlsn = startlsn + data.len() as u64;
trace!("received XLogData between {} and {}", startlsn, endlsn);
info!("received XLogData between {} and {}", startlsn, endlsn);
waldecoder.feed_bytes(data);