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

Merge fixes

Browse Source
This commit is contained in:
Heikki Linnakangas
2021-08-01 13:44:02 +03:00
parent 2b080b49c4
commit fbff37a64c
5 changed files with 31 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files

31
pageserver/src/branches.rs
View File

@@ -25,7 +25,7 @@ use crate::page_cache;
use crate::restore_local_repo;
use crate::walredo::WalRedoManager;
use crate::ZTenantId;
use crate::{repository::Repository, PageServerConf, ZTimelineId};
use crate::{repository::Repository, PageServerConf, RepositoryFormat, ZTimelineId};
#[derive(Serialize, Deserialize, Clone)]
pub struct BranchInfo {
@@ -73,8 +73,8 @@ pub fn init_pageserver(
pub fn create_repo(
conf: &'static PageServerConf,
tenantid: ZTenantId,
wal_redo_manager: Arc<dyn WalRedoManager>,
) -> Result<ObjectRepository> {
wal_redo_manager: Arc<dyn WalRedoManager + Send + Sync>,
) -> Result<Arc<dyn Repository>> {
let repo_dir = conf.tenant_path(&tenantid);
if repo_dir.exists() {
bail!("repo for {} already exists", tenantid)
@@ -104,19 +104,28 @@ pub fn create_repo(
// and we failed to run initdb again in the same directory. This has been solved for the
// rapid init+start case now, but the general race condition remains if you restart the
// server quickly.
let storage = crate::rocksdb_storage::RocksObjectStore::create(conf, &tenantid)?;
let repo: Arc<dyn Repository + Sync + Send> = match conf.repository_format {
RepositoryFormat::Layered => Arc::new(
crate::layered_repository::LayeredRepository::new(conf,
wal_redo_manager,
tenantid
)),
RepositoryFormat::RocksDb => {
let obj_store = crate::rocksdb_storage::RocksObjectStore::open(conf, &tenantid).unwrap();
let repo = crate::object_repository::ObjectRepository::new(
conf,
std::sync::Arc::new(storage),
wal_redo_manager,
tenantid,
);
Arc::new(ObjectRepository::new(
conf,
Arc::new(obj_store),
wal_redo_manager,
tenantid
))
}
};
// Load data into pageserver
// TODO To implement zenith import we need to
// move data loading out of create_repo()
bootstrap_timeline(conf, tenantid, tli, &repo)?;
bootstrap_timeline(conf, tenantid, tli, &*repo)?;
Ok(repo)
}

4
pageserver/src/layered_repository.rs
View File

@@ -201,6 +201,8 @@ impl LayeredRepository {
let path = conf.timeline_path(&timelineid, &tenantid).join("metadata");
let mut file = File::create(&path)?;
info!("saving metadata {}", path.display());
file.write_all(&TimelineMetadata::ser(data)?)?;
Ok(())
@@ -256,7 +258,7 @@ impl LayeredRepository {
// Remember timelineid and its last_record_lsn for each timeline
let mut timelines: Vec<(ZTimelineId, Lsn)> = Vec::new();
let timelines_path = conf.workdir.join("timelines");
let timelines_path = conf.timelines_path(&tenantid);
for direntry in fs::read_dir(timelines_path)? {
let direntry = direntry?;
if let Some(fname) = direntry.file_name().to_str() {

3
pageserver/src/page_cache.rs
View File

@@ -24,7 +24,6 @@ lazy_static! {
pub fn init(conf: &'static PageServerConf) {
let mut m = REPOSITORY.lock().unwrap();
for dir_entry in fs::read_dir(conf.tenants_path()).unwrap() {
let tenantid =
ZTenantId::from_str(dir_entry.unwrap().file_name().to_str().unwrap()).unwrap();
@@ -68,7 +67,7 @@ pub fn create_repository_for_tenant(
let wal_redo_manager = Arc::new(PostgresRedoManager::new(conf, tenantid));
let repo = branches::create_repo(conf, tenantid, wal_redo_manager)?;
m.insert(tenantid, Arc::new(repo));
m.insert(tenantid, repo);
Ok(())
}

2
pageserver/src/repository.rs
View File

@@ -362,7 +362,7 @@ mod tests {
let repo: Box<dyn Repository + Sync + Send> = match conf.repository_format {
RepositoryFormat::Layered => {
Box::new(LayeredRepository::new(conf, Arc::new(walredo_mgr)))
Box::new(LayeredRepository::new(conf, Arc::new(walredo_mgr), tenantid))
}
RepositoryFormat::RocksDb => {
let obj_store = RocksObjectStore::create(conf, &tenantid)?;

12
test_runner/batch_others/test_snapfiles_gc.py
View File

@@ -42,7 +42,7 @@ def test_snapfiles_gc(zenith_cli, pageserver, postgres, pg_bin):
cur.execute("VACUUM")
print("Running GC before test")
pscur.execute(f"do_gc {timeline} 0")
pscur.execute(f"do_gc {pageserver.initial_tenant} {timeline} 0")
row = pscur.fetchone()
print_gc_result(row);
# remember the number of files
@@ -54,7 +54,7 @@ def test_snapfiles_gc(zenith_cli, pageserver, postgres, pg_bin):
snapshot_relfiles_remain += 1;
print("Inserting one row and running GC")
cur.execute("INSERT INTO foo VALUES (1)")
pscur.execute(f"do_gc {timeline} 0")
pscur.execute(f"do_gc {pageserver.initial_tenant} {timeline} 0")
row = pscur.fetchone()
print_gc_result(row);
assert row['snapshot_relfiles_total'] == snapshot_relfiles_remain
@@ -68,7 +68,7 @@ def test_snapfiles_gc(zenith_cli, pageserver, postgres, pg_bin):
cur.execute("INSERT INTO foo VALUES (2)")
cur.execute("INSERT INTO foo VALUES (3)")
pscur.execute(f"do_gc {timeline} 0")
pscur.execute(f"do_gc {pageserver.initial_tenant} {timeline} 0")
row = pscur.fetchone()
print_gc_result(row);
assert row['snapshot_relfiles_total'] == snapshot_relfiles_remain + 1
@@ -80,7 +80,7 @@ def test_snapfiles_gc(zenith_cli, pageserver, postgres, pg_bin):
cur.execute("INSERT INTO foo VALUES (2)")
cur.execute("INSERT INTO foo VALUES (3)")
pscur.execute(f"do_gc {timeline} 0")
pscur.execute(f"do_gc {pageserver.initial_tenant} {timeline} 0")
row = pscur.fetchone()
print_gc_result(row);
assert row['snapshot_relfiles_total'] == snapshot_relfiles_remain + 1
@@ -89,7 +89,7 @@ def test_snapfiles_gc(zenith_cli, pageserver, postgres, pg_bin):
# Run GC again, with no changes in the database. Should not remove anything.
print("Run GC again, with nothing to do")
pscur.execute(f"do_gc {timeline} 0")
pscur.execute(f"do_gc {pageserver.initial_tenant} {timeline} 0")
row = pscur.fetchone()
print_gc_result(row);
assert row['snapshot_relfiles_total'] == snapshot_relfiles_remain
@@ -102,7 +102,7 @@ def test_snapfiles_gc(zenith_cli, pageserver, postgres, pg_bin):
print("Drop table and run GC again");
cur.execute("DROP TABLE foo")
pscur.execute(f"do_gc {timeline} 0")
pscur.execute(f"do_gc {pageserver.initial_tenant} {timeline} 0")
row = pscur.fetchone()
print_gc_result(row);