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

Handle tablespaces gracefully.

Browse Source 
We don't really support tablespaces. But this makes the 'tablespace'
Postgres regression test pass, like it did previously.
This commit is contained in:
Heikki Linnakangas
2022-03-10 22:56:37 +02:00
parent 0e3512aad0
commit a726b555fb
5 changed files with 114 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files

55
pageserver/src/basebackup.rs
View File

@@ -125,8 +125,10 @@ impl<'a> Basebackup<'a> {
self.add_slru_segment(kind, segno)?;
}
}
for (spcnode, dbnode) in self.timeline.list_relmap_files(self.lsn)? {
self.add_relmap_file(spcnode, dbnode)?;
// Create tablespace directories
for ((spcnode, dbnode), has_relmap_file) in self.timeline.list_dbdirs(self.lsn)? {
self.add_dbdir(spcnode, dbnode, has_relmap_file)?;
}
for xid in self.timeline.list_twophase_files(self.lsn)? {
self.add_twophase_file(xid)?;
@@ -165,12 +167,26 @@ impl<'a> Basebackup<'a> {
}
//
// Extract pg_filenode.map files from repository
// Along with them also send PG_VERSION for each database.
// Include database/tablespace directories.
//
fn add_relmap_file(&mut self, spcnode: u32, dbnode: u32) -> anyhow::Result<()> {
let img = self.timeline.get_relmap_file(spcnode, dbnode, self.lsn)?;
let path = if spcnode == pg_constants::GLOBALTABLESPACE_OID {
// Each directory contains a PG_VERSION file, and the default database
// directories also contain pg_filenode.map files.
//
fn add_dbdir(
&mut self,
spcnode: u32,
dbnode: u32,
has_relmap_file: bool,
) -> anyhow::Result<()> {
let relmap_img = if has_relmap_file {
let img = self.timeline.get_relmap_file(spcnode, dbnode, self.lsn)?;
assert!(img.len() == 512);
Some(img)
} else {
None
};
if spcnode == pg_constants::GLOBALTABLESPACE_OID {
let version_bytes = pg_constants::PG_MAJORVERSION.as_bytes();
let header = new_tar_header("PG_VERSION", version_bytes.len() as u64)?;
self.ar.append(&header, version_bytes)?;
@@ -178,7 +194,13 @@ impl<'a> Basebackup<'a> {
let header = new_tar_header("global/PG_VERSION", version_bytes.len() as u64)?;
self.ar.append(&header, version_bytes)?;
String::from("global/pg_filenode.map") // filenode map for global tablespace
if let Some(img) = relmap_img {
// filenode map for global tablespace
let header = new_tar_header("global/pg_filenode.map", img.len() as u64)?;
self.ar.append(&header, &img[..])?;
} else {
warn!("global/pg_filenode.map is missing");
}
} else {
// User defined tablespaces are not supported
assert!(spcnode == pg_constants::DEFAULTTABLESPACE_OID);
@@ -188,16 +210,17 @@ impl<'a> Basebackup<'a> {
let header = new_tar_header_dir(&path)?;
self.ar.append(&header, &mut io::empty())?;
let dst_path = format!("base/{}/PG_VERSION", dbnode);
let version_bytes = pg_constants::PG_MAJORVERSION.as_bytes();
let header = new_tar_header(&dst_path, version_bytes.len() as u64)?;
self.ar.append(&header, version_bytes)?;
if let Some(img) = relmap_img {
let dst_path = format!("base/{}/PG_VERSION", dbnode);
let version_bytes = pg_constants::PG_MAJORVERSION.as_bytes();
let header = new_tar_header(&dst_path, version_bytes.len() as u64)?;
self.ar.append(&header, version_bytes)?;
format!("base/{}/pg_filenode.map", dbnode)
let relmap_path = format!("base/{}/pg_filenode.map", dbnode);
let header = new_tar_header(&relmap_path, img.len() as u64)?;
self.ar.append(&header, &img[..])?;
}
};
assert!(img.len() == 512);
let header = new_tar_header(&path, img.len() as u64)?;
self.ar.append(&header, &img[..])?;
Ok(())
}

3
pageserver/src/import_datadir.rs
View File

@@ -40,7 +40,6 @@ pub fn import_timeline_from_postgres_datadir<R: Repository>(
// Scan 'global'
let mut relfiles: Vec<PathBuf> = Vec::new();
writer.put_dbdir_creation(pg_constants::GLOBALTABLESPACE_OID, 0)?;
for direntry in fs::read_dir(path.join("global"))? {
let direntry = direntry?;
match direntry.file_name().to_str() {
@@ -85,7 +84,7 @@ pub fn import_timeline_from_postgres_datadir<R: Repository>(
None => continue,
Some("PG_VERSION") => {
writer.put_dbdir_creation(pg_constants::DEFAULTTABLESPACE_OID, dboid)?;
//writer.put_dbdir_creation(pg_constants::DEFAULTTABLESPACE_OID, dboid)?;
}
Some("pg_filenode.map") => import_relmap_file(
&mut writer,

3
pageserver/src/layered_repository.rs
View File

@@ -1071,7 +1071,8 @@ impl LayeredTimeline {
if Lsn(cont_lsn.0 - 1) <= timeline.ancestor_lsn {
trace!(
"going into ancestor {}, cont_lsn is {}",
timeline.ancestor_lsn, cont_lsn
timeline.ancestor_lsn,
cont_lsn
);
let ancestor = timeline.get_ancestor_timeline()?;
timeline_owned = ancestor;

89
pageserver/src/pgdatadir_mapping.rs
View File

@@ -39,8 +39,8 @@ where
#[derive(Debug, Serialize, Deserialize)]
pub struct DbDirectory {
// (spcnode, dbnode)
dbs: HashSet<(Oid, Oid)>,
// (spcnode, dbnode) -> (do relmapper and PG_VERSION files exist)
dbdirs: HashMap<(Oid, Oid), bool>,
}
#[derive(Debug, Serialize, Deserialize)]
@@ -210,12 +210,12 @@ impl<R: Repository> DatadirTimeline<R> {
Ok(buf)
}
pub fn list_relmap_files(&self, lsn: Lsn) -> Result<HashSet<(Oid, Oid)>> {
pub fn list_dbdirs(&self, lsn: Lsn) -> Result<HashMap<(Oid, Oid), bool>> {
// fetch directory entry
let buf = self.tline.get(DBDIR_KEY, lsn)?;
let dir = DbDirectory::des(&buf)?;
Ok(dir.dbs)
Ok(dir.dbdirs)
}
pub fn get_twophase_file(&self, xid: TransactionId, lsn: Lsn) -> Result<Bytes> {
@@ -324,8 +324,8 @@ impl<R: Repository> DatadirTimeline<R> {
let dbdir = DbDirectory::des(&buf)?;
let mut total_size: usize = 0;
for (spcnode, dbnode) in dbdir.dbs {
for rel in self.list_rels(spcnode, dbnode, lsn)? {
for (spcnode, dbnode) in dbdir.dbdirs.keys() {
for rel in self.list_rels(*spcnode, *dbnode, lsn)? {
let relsize_key = rel_size_to_key(rel);
let mut buf = self.tline.get(relsize_key, lsn)?;
let relsize = buf.get_u32_le();
@@ -347,7 +347,7 @@ impl<R: Repository> DatadirTimeline<R> {
let buf = self.tline.get(DBDIR_KEY, lsn)?;
let dbdir = DbDirectory::des(&buf)?;
let mut dbs: Vec<(Oid, Oid)> = dbdir.dbs.iter().cloned().collect();
let mut dbs: Vec<(Oid, Oid)> = dbdir.dbdirs.keys().cloned().collect();
dbs.sort_unstable();
for (spcnode, dbnode) in dbs {
result.add_key(relmap_file_key(spcnode, dbnode));
@@ -432,7 +432,7 @@ impl<'a, R: Repository> std::ops::Deref for DatadirTimelineWriter<'a, R> {
impl<'a, R: Repository> DatadirTimelineWriter<'a, R> {
pub fn init_empty(&mut self) -> Result<()> {
let buf = DbDirectory::ser(&DbDirectory {
dbs: HashSet::new(),
dbdirs: HashMap::new(),
})?;
self.put(DBDIR_KEY, Value::Image(buf.into()));
@@ -512,11 +512,26 @@ impl<'a, R: Repository> DatadirTimelineWriter<'a, R> {
pub fn put_relmap_file(&mut self, spcnode: Oid, dbnode: Oid, img: Bytes) -> Result<()> {
// Add it to the directory (if it doesn't exist already)
let buf = self.get(DBDIR_KEY)?;
let mut dir = DbDirectory::des(&buf)?;
if dir.dbs.insert((spcnode, dbnode)) {
let buf = DbDirectory::ser(&dir)?;
let mut dbdir = DbDirectory::des(&buf)?;
let r = dbdir.dbdirs.insert((spcnode, dbnode), true);
if r == None || r == Some(false) {
// The dbdir entry didn't exist, or it contained a
// 'false'. The 'insert' call already updated it with
// 'true', now write the updated 'dbdirs' map back.
let buf = DbDirectory::ser(&dbdir)?;
self.put(DBDIR_KEY, Value::Image(buf.into()));
}
if r == None {
// Create RelDirectory
let buf = RelDirectory::ser(&RelDirectory {
rels: HashSet::new(),
})?;
self.put(
rel_dir_to_key(spcnode, dbnode),
Value::Image(Bytes::from(buf)),
);
}
self.put(relmap_file_key(spcnode, dbnode), Value::Image(img));
Ok(())
@@ -548,23 +563,11 @@ impl<'a, R: Repository> DatadirTimelineWriter<'a, R> {
Ok(())
}
pub fn put_dbdir_creation(&mut self, spcnode: Oid, dbnode: Oid) -> Result<()> {
// Create RelDirectory
let dir_key = rel_dir_to_key(spcnode, dbnode);
let dir = RelDirectory {
rels: HashSet::new(),
};
let buf: Bytes = RelDirectory::ser(&dir)?.into();
self.put(dir_key, Value::Image(buf));
Ok(())
}
pub fn drop_dbdir(&mut self, spcnode: Oid, dbnode: Oid) -> Result<()> {
// Remove entry from dbdir
let buf = self.get(DBDIR_KEY)?;
let mut dir = DbDirectory::des(&buf)?;
if dir.dbs.remove(&(spcnode, dbnode)) {
if dir.dbdirs.remove(&(spcnode, dbnode)).is_some() {
let buf = DbDirectory::ser(&dir)?;
self.put(DBDIR_KEY, Value::Image(buf.into()));
} else {
@@ -594,15 +597,36 @@ impl<'a, R: Repository> DatadirTimelineWriter<'a, R> {
// - update relish header with size
pub fn put_rel_creation(&mut self, rel: RelTag, nblocks: BlockNumber) -> Result<()> {
// Add it to the directory entry
let dir_key = rel_dir_to_key(rel.spcnode, rel.dbnode);
let buf = self.get(dir_key)?;
let mut dir = RelDirectory::des(&buf)?;
// It's possible that this is the first rel for this db in this tablespace.
// Create the reldir entry for it if so.
let buf = self.get(DBDIR_KEY)?;
let mut dbdir = DbDirectory::des(&buf)?;
if !dir.rels.insert((rel.relnode, rel.forknum)) {
let rel_dir_key = rel_dir_to_key(rel.spcnode, rel.dbnode);
let mut rel_dir;
if dbdir.dbdirs.get(&(rel.spcnode, rel.dbnode)).is_none() {
// update dbdir
dbdir.dbdirs.insert((rel.spcnode, rel.dbnode), false);
let buf = DbDirectory::ser(&dbdir)?;
self.put(DBDIR_KEY, Value::Image(buf.into()));
// Create RelDirectory
rel_dir = RelDirectory {
rels: HashSet::new(),
};
} else {
let buf = self.get(rel_dir_key)?;
rel_dir = RelDirectory::des(&buf)?;
}
// Add it to the directory entry
if !rel_dir.rels.insert((rel.relnode, rel.forknum)) {
bail!("rel {} already exists", rel);
}
self.put(dir_key, Value::Image(Bytes::from(RelDirectory::ser(&dir)?)));
self.put(
rel_dir_key,
Value::Image(Bytes::from(RelDirectory::ser(&rel_dir)?)),
);
// Put size
let size_key = rel_size_to_key(rel);
@@ -611,7 +635,8 @@ impl<'a, R: Repository> DatadirTimelineWriter<'a, R> {
self.pending_nblocks += nblocks as isize;
// even if nblocks > 0, we don't insert any actual blocks here
// Even if nblocks > 0, we don't insert any actual blocks here. That's up to the
// caller.
Ok(())
}
@@ -1155,8 +1180,6 @@ pub fn create_test_timeline<R: Repository>(
let mut writer = tline.begin_record(Lsn(8));
writer.init_empty()?;
writer.put_dbdir_creation(0, 111)?;
writer.finish()?;
Ok(Arc::new(tline))
}

22
pageserver/src/walingest.rs
View File

@@ -461,7 +461,9 @@ impl<'a, R: Repository> WalIngest<'a, R> {
debug!("ingest_xlog_dbase_create: {} rels", rels.len());
timeline.put_dbdir_creation(tablespace_id, db_id)?;
// Copy relfilemap
let filemap = timeline.get_relmap_file(src_tablespace_id, src_db_id, req_lsn)?;
timeline.put_relmap_file(tablespace_id, db_id, filemap)?;
let mut num_rels_copied = 0;
let mut num_blocks_copied = 0;
@@ -492,10 +494,6 @@ impl<'a, R: Repository> WalIngest<'a, R> {
num_rels_copied += 1;
}
// Copy relfilemap
let filemap = timeline.get_relmap_file(src_tablespace_id, src_db_id, req_lsn)?;
timeline.put_relmap_file(tablespace_id, db_id, filemap)?;
info!(
"Created database {}/{}, copied {} blocks in {} rels",
tablespace_id, db_id, num_blocks_copied, num_rels_copied
@@ -514,7 +512,7 @@ impl<'a, R: Repository> WalIngest<'a, R> {
relnode: rec.rnode.relnode,
forknum: rec.forknum,
};
writer.put_rel_creation(rel, 0)?;
self.put_rel_creation(writer, rel)?;
Ok(())
}
@@ -833,6 +831,16 @@ impl<'a, R: Repository> WalIngest<'a, R> {
Ok(())
}
fn put_rel_creation(
&mut self,
writer: &mut DatadirTimelineWriter<R>,
rel: RelTag,
) -> Result<()> {
self.relsize_cache.insert(rel, 0);
writer.put_rel_creation(rel, 0)?;
Ok(())
}
fn put_rel_page_image(
&mut self,
writer: &mut DatadirTimelineWriter<R>,
@@ -999,6 +1007,7 @@ mod tests {
fn init_walingest_test<R: Repository>(tline: &DatadirTimeline<R>) -> Result<WalIngest<R>> {
let mut writer = tline.begin_record(Lsn(0x10));
writer.put_checkpoint(ZERO_CHECKPOINT.clone())?;
writer.put_relmap_file(0, 111, Bytes::from(""))?; // dummy relmapper file
writer.finish()?;
let walingest = WalIngest::new(tline, Lsn(0x10))?;
@@ -1012,6 +1021,7 @@ mod tests {
let mut walingest = init_walingest_test(&tline)?;
let mut writer = tline.begin_record(Lsn(0x20));
walingest.put_rel_creation(&mut writer, TESTREL_A)?;
walingest.put_rel_page_image(&mut writer, TESTREL_A, 0, TEST_IMG("foo blk 0 at 2"))?;
writer.finish()?;
let mut writer = tline.begin_record(Lsn(0x30));