DatabaseSource::Auto (#9500)

* implement "auto" database backend in client/db, in progress, #9201

* move fn supports_ref_counting from DatabaseSource enum to Database trait to make it work correctly for all types of dbs

* update kvdb_rocksdb to 0.13 and use it's new config feature  to properly auto start existing database

* tests for auto database reopening

* introduce OpenDbError to cleanup opening database error handling and handle case when database is not enabled at the compile time

* cargo fmt strings again

* cargo fmt strings again

* rename DataSettingsSrc to fix test compilation

* fix the call to the new kvdb-rocksdb interdace in tests to fix compilation

* simplify OpenDbError and make it compile even when paritydb and rocksdb are disabled

* cargo fmt

* fix compilation without flag with-parity-db

* fix unused var compilation warning

* support different paths for rocksdb and paritydb in DatabaseSouce::Auto

* support "auto" database option in substrate cli

* enable Lz4 compression for some of the parity-db colums as per review suggestion

* applied review suggestions

substrate/client/db/src/lib.rs

@@ -297,7 +297,7 @@ pub struct DatabaseSettings {
 	/// State pruning mode.
 	pub state_pruning: PruningMode,
 	/// Where to find the database.
-	pub source: DatabaseSettingsSrc,
+	pub source: DatabaseSource,
 	/// Block pruning mode.
 	pub keep_blocks: KeepBlocks,
 	/// Block body/Transaction storage scheme.
@@ -325,7 +325,17 @@ pub enum TransactionStorageMode {
 
 /// Where to find the database..
 #[derive(Debug, Clone)]
-pub enum DatabaseSettingsSrc {
+pub enum DatabaseSource {
+	/// Check given path, and see if there is an existing database there. If it's either `RocksDb`
+	/// or `ParityDb`, use it. If there is none, create a new instance of `ParityDb`.
+	Auto {
+		/// Path to the paritydb database.
+		paritydb_path: PathBuf,
+		/// Path to the rocksdb database.
+		rocksdb_path: PathBuf,
+		/// Cache size in MiB. Used only by `RocksDb` variant of `DatabaseSource`.
+		cache_size: usize,
+	},
 	/// Load a RocksDB database from a given path. Recommended for most uses.
 	RocksDb {
 		/// Path to the database.
@@ -344,27 +354,28 @@ pub enum DatabaseSettingsSrc {
 	Custom(Arc<dyn Database<DbHash>>),
 }
 
-impl DatabaseSettingsSrc {
+impl DatabaseSource {
 	/// Return dabase path for databases that are on the disk.
 	pub fn path(&self) -> Option<&Path> {
 		match self {
-			DatabaseSettingsSrc::RocksDb { path, .. } => Some(path.as_path()),
-			DatabaseSettingsSrc::ParityDb { path, .. } => Some(path.as_path()),
-			DatabaseSettingsSrc::Custom(_) => None,
+			// as per https://github.com/paritytech/substrate/pull/9500#discussion_r684312550
+			//
+			// IIUC this is needed for polkadot to create its own dbs, so until it can use parity db
+			// I would think rocksdb, but later parity-db.
+			DatabaseSource::Auto { paritydb_path, .. } => Some(&paritydb_path),
+			DatabaseSource::RocksDb { path, .. } | DatabaseSource::ParityDb { path } => Some(&path),
+			DatabaseSource::Custom(..) => None,
 		}
 	}
-	/// Check if database supports internal ref counting for state data.
-	pub fn supports_ref_counting(&self) -> bool {
-		matches!(self, DatabaseSettingsSrc::ParityDb { .. })
-	}
 }
 
-impl std::fmt::Display for DatabaseSettingsSrc {
+impl std::fmt::Display for DatabaseSource {
 	fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
 		let name = match self {
-			DatabaseSettingsSrc::RocksDb { .. } => "RocksDb",
-			DatabaseSettingsSrc::ParityDb { .. } => "ParityDb",
-			DatabaseSettingsSrc::Custom(_) => "Custom",
+			DatabaseSource::Auto { .. } => "Auto",
+			DatabaseSource::RocksDb { .. } => "RocksDb",
+			DatabaseSource::ParityDb { .. } => "ParityDb",
+			DatabaseSource::Custom(_) => "Custom",
 		};
 		write!(f, "{}", name)
 	}
@@ -1106,7 +1117,7 @@ impl<Block: BlockT> Backend<Block> {
 			state_cache_size: 16777216,
 			state_cache_child_ratio: Some((50, 100)),
 			state_pruning: PruningMode::keep_blocks(keep_blocks),
-			source: DatabaseSettingsSrc::Custom(db),
+			source: DatabaseSource::Custom(db),
 			keep_blocks: KeepBlocks::Some(keep_blocks),
 			transaction_storage,
 		};
@@ -1125,15 +1136,12 @@ impl<Block: BlockT> Backend<Block> {
 		let map_e = |e: sc_state_db::Error<io::Error>| sp_blockchain::Error::from_state_db(e);
 		let state_db: StateDb<_, _> = StateDb::new(
 			config.state_pruning.clone(),
-			!config.source.supports_ref_counting(),
+			!db.supports_ref_counting(),
 			&StateMetaDb(&*db),
 		)
 		.map_err(map_e)?;
-		let storage_db = StorageDb {
-			db: db.clone(),
-			state_db,
-			prefix_keys: !config.source.supports_ref_counting(),
-		};
+		let storage_db =
+			StorageDb { db: db.clone(), state_db, prefix_keys: !db.supports_ref_counting() };
 		let offchain_storage = offchain::LocalStorage::new(db.clone());
 		let changes_tries_storage = DbChangesTrieStorage::new(
 			db,
@@ -2516,7 +2524,7 @@ pub(crate) mod tests {
 				state_cache_size: 16777216,
 				state_cache_child_ratio: Some((50, 100)),
 				state_pruning: PruningMode::keep_blocks(1),
-				source: DatabaseSettingsSrc::Custom(backing),
+				source: DatabaseSource::Custom(backing),
 				keep_blocks: KeepBlocks::All,
 				transaction_storage: TransactionStorageMode::BlockBody,
 			},

substrate/client/db/src/parity_db.rs

@@ -16,7 +16,7 @@
 // You should have received a copy of the GNU General Public License
 // along with this program. If not, see <https://www.gnu.org/licenses/>.
 use crate::{
-	columns,
+	columns, light,
 	utils::{DatabaseType, NUM_COLUMNS},
 };
 /// A `Database` adapter for parity-db.
@@ -37,16 +37,42 @@ fn handle_err<T>(result: parity_db::Result<T>) -> T {
 pub fn open<H: Clone + AsRef<[u8]>>(
 	path: &std::path::Path,
 	db_type: DatabaseType,
+	create: bool,
 ) -> parity_db::Result<std::sync::Arc<dyn Database<H>>> {
 	let mut config = parity_db::Options::with_columns(path, NUM_COLUMNS as u8);
-	config.sync = true; // Flush each commit
-	if db_type == DatabaseType::Full {
-		let mut state_col = &mut config.columns[columns::STATE as usize];
-		state_col.ref_counted = true;
-		state_col.preimage = true;
-		state_col.uniform = true;
+
+	match db_type {
+		DatabaseType::Full => {
+			let indexes = [
+				columns::STATE,
+				columns::HEADER,
+				columns::BODY,
+				columns::TRANSACTION,
+				columns::JUSTIFICATIONS,
+			];
+
+			for i in indexes {
+				let mut column = &mut config.columns[i as usize];
+				column.compression = parity_db::CompressionType::Lz4;
+			}
+
+			let mut state_col = &mut config.columns[columns::STATE as usize];
+			state_col.ref_counted = true;
+			state_col.preimage = true;
+			state_col.uniform = true;
+		},
+		DatabaseType::Light => {
+			config.columns[light::columns::HEADER as usize].compression =
+				parity_db::CompressionType::Lz4;
+		},
 	}
-	let db = parity_db::Db::open(&config)?;
+
+	let db = if create {
+		parity_db::Db::open_or_create(&config)?
+	} else {
+		parity_db::Db::open(&config)?
+	};
+
 	Ok(std::sync::Arc::new(DbAdapter(db)))
 }
 
@@ -72,4 +98,8 @@ impl<H: Clone + AsRef<[u8]>> Database<H> for DbAdapter {
 	fn value_size(&self, col: ColumnId, key: &[u8]) -> Option<usize> {
 		handle_err(self.0.get_size(col as u8, key)).map(|s| s as usize)
 	}
+
+	fn supports_ref_counting(&self) -> bool {
+		true
+	}
 }

substrate/client/db/src/upgrade.rs

@@ -19,8 +19,8 @@
 //! Database upgrade logic.
 
 use std::{
-	fs,
-	io::{ErrorKind, Read, Write},
+	fmt, fs,
+	io::{self, ErrorKind, Read, Write},
 	path::{Path, PathBuf},
 };
 
@@ -39,61 +39,79 @@ const CURRENT_VERSION: u32 = 3;
 const V1_NUM_COLUMNS: u32 = 11;
 const V2_NUM_COLUMNS: u32 = 12;
 
-/// Upgrade database to current version.
-pub fn upgrade_db<Block: BlockT>(
-	db_path: &Path,
-	db_type: DatabaseType,
-) -> sp_blockchain::Result<()> {
-	let is_empty = db_path.read_dir().map_or(true, |mut d| d.next().is_none());
-	if !is_empty {
-		let db_version = current_version(db_path)?;
-		match db_version {
-			0 => Err(sp_blockchain::Error::Backend(format!(
-				"Unsupported database version: {}",
-				db_version
-			)))?,
-			1 => {
-				migrate_1_to_2::<Block>(db_path, db_type)?;
-				migrate_2_to_3::<Block>(db_path, db_type)?
-			},
-			2 => migrate_2_to_3::<Block>(db_path, db_type)?,
-			CURRENT_VERSION => (),
-			_ => Err(sp_blockchain::Error::Backend(format!(
-				"Future database version: {}",
-				db_version
-			)))?,
+/// Database upgrade errors.
+#[derive(Debug)]
+pub enum UpgradeError {
+	/// Database version cannot be read from existing db_version file.
+	UnknownDatabaseVersion,
+	/// Missing database version file.
+	MissingDatabaseVersionFile,
+	/// Database version no longer supported.
+	UnsupportedVersion(u32),
+	/// Database version comes from future version of the client.
+	FutureDatabaseVersion(u32),
+	/// Invalid justification block.
+	DecodingJustificationBlock,
+	/// Common io error.
+	Io(io::Error),
+}
+
+pub type UpgradeResult<T> = Result<T, UpgradeError>;
+
+impl From<io::Error> for UpgradeError {
+	fn from(err: io::Error) -> Self {
+		UpgradeError::Io(err)
+	}
+}
+
+impl fmt::Display for UpgradeError {
+	fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+		match self {
+			UpgradeError::UnknownDatabaseVersion =>
+				write!(f, "Database version cannot be read from exisiting db_version file"),
+			UpgradeError::MissingDatabaseVersionFile => write!(f, "Missing database version file"),
+			UpgradeError::UnsupportedVersion(version) =>
+				write!(f, "Database version no longer supported: {}", version),
+			UpgradeError::FutureDatabaseVersion(version) =>
+				write!(f, "Database version comes from future version of the client: {}", version),
+			UpgradeError::DecodingJustificationBlock =>
+				write!(f, "Decodoning justification block failed"),
+			UpgradeError::Io(err) => write!(f, "Io error: {}", err),
 		}
 	}
+}
 
-	update_version(db_path)
+/// Upgrade database to current version.
+pub fn upgrade_db<Block: BlockT>(db_path: &Path, db_type: DatabaseType) -> UpgradeResult<()> {
+	let db_version = current_version(db_path)?;
+	match db_version {
+		0 => return Err(UpgradeError::UnsupportedVersion(db_version)),
+		1 => {
+			migrate_1_to_2::<Block>(db_path, db_type)?;
+			migrate_2_to_3::<Block>(db_path, db_type)?
+		},
+		2 => migrate_2_to_3::<Block>(db_path, db_type)?,
+		CURRENT_VERSION => (),
+		_ => return Err(UpgradeError::FutureDatabaseVersion(db_version)),
+	}
+	update_version(db_path)?;
+	Ok(())
 }
 
 /// Migration from version1 to version2:
 /// 1) the number of columns has changed from 11 to 12;
 /// 2) transactions column is added;
-fn migrate_1_to_2<Block: BlockT>(
-	db_path: &Path,
-	_db_type: DatabaseType,
-) -> sp_blockchain::Result<()> {
-	let db_path = db_path
-		.to_str()
-		.ok_or_else(|| sp_blockchain::Error::Backend("Invalid database path".into()))?;
+fn migrate_1_to_2<Block: BlockT>(db_path: &Path, _db_type: DatabaseType) -> UpgradeResult<()> {
 	let db_cfg = DatabaseConfig::with_columns(V1_NUM_COLUMNS);
-	let db = Database::open(&db_cfg, db_path).map_err(db_err)?;
-	db.add_column().map_err(db_err)
+	let db = Database::open(&db_cfg, db_path)?;
+	db.add_column().map_err(Into::into)
 }
 
 /// Migration from version2 to version3:
 /// - The format of the stored Justification changed to support multiple Justifications.
-fn migrate_2_to_3<Block: BlockT>(
-	db_path: &Path,
-	_db_type: DatabaseType,
-) -> sp_blockchain::Result<()> {
-	let db_path = db_path
-		.to_str()
-		.ok_or_else(|| sp_blockchain::Error::Backend("Invalid database path".into()))?;
+fn migrate_2_to_3<Block: BlockT>(db_path: &Path, _db_type: DatabaseType) -> UpgradeResult<()> {
 	let db_cfg = DatabaseConfig::with_columns(V2_NUM_COLUMNS);
-	let db = Database::open(&db_cfg, db_path).map_err(db_err)?;
+	let db = Database::open(&db_cfg, db_path)?;
 
 	// Get all the keys we need to update
 	let keys: Vec<_> = db.iter(columns::JUSTIFICATIONS).map(|entry| entry.0).collect();
@@ -101,49 +119,43 @@ fn migrate_2_to_3<Block: BlockT>(
 	// Read and update each entry
 	let mut transaction = db.transaction();
 	for key in keys {
-		if let Some(justification) = db.get(columns::JUSTIFICATIONS, &key).map_err(db_err)? {
+		if let Some(justification) = db.get(columns::JUSTIFICATIONS, &key)? {
 			// Tag each justification with the hardcoded ID for GRANDPA to avoid the dependency on
 			// the GRANDPA crate.
 			// NOTE: when storing justifications the previous API would get a `Vec<u8>` and still
 			// call encode on it.
 			let justification = Vec::<u8>::decode(&mut &justification[..])
-				.map_err(|_| sp_blockchain::Error::Backend("Invalid justification blob".into()))?;
+				.map_err(|_| UpgradeError::DecodingJustificationBlock)?;
 			let justifications = sp_runtime::Justifications::from((*b"FRNK", justification));
 			transaction.put_vec(columns::JUSTIFICATIONS, &key, justifications.encode());
 		}
 	}
-	db.write(transaction).map_err(db_err)?;
+	db.write(transaction)?;
 
 	Ok(())
 }
 
 /// Reads current database version from the file at given path.
 /// If the file does not exist returns 0.
-fn current_version(path: &Path) -> sp_blockchain::Result<u32> {
-	let unknown_version_err = || sp_blockchain::Error::Backend("Unknown database version".into());
-
+fn current_version(path: &Path) -> UpgradeResult<u32> {
 	match fs::File::open(version_file_path(path)) {
-		Err(ref err) if err.kind() == ErrorKind::NotFound => Ok(0),
-		Err(_) => Err(unknown_version_err()),
+		Err(ref err) if err.kind() == ErrorKind::NotFound =>
+			Err(UpgradeError::MissingDatabaseVersionFile),
+		Err(_) => Err(UpgradeError::UnknownDatabaseVersion),
 		Ok(mut file) => {
 			let mut s = String::new();
-			file.read_to_string(&mut s).map_err(|_| unknown_version_err())?;
-			u32::from_str_radix(&s, 10).map_err(|_| unknown_version_err())
+			file.read_to_string(&mut s).map_err(|_| UpgradeError::UnknownDatabaseVersion)?;
+			u32::from_str_radix(&s, 10).map_err(|_| UpgradeError::UnknownDatabaseVersion)
 		},
 	}
 }
 
-/// Maps database error to client error
-fn db_err(err: std::io::Error) -> sp_blockchain::Error {
-	sp_blockchain::Error::Backend(format!("{}", err))
-}
-
 /// Writes current database version to the file.
 /// Creates a new file if the version file does not exist yet.
-fn update_version(path: &Path) -> sp_blockchain::Result<()> {
-	fs::create_dir_all(path).map_err(db_err)?;
-	let mut file = fs::File::create(version_file_path(path)).map_err(db_err)?;
-	file.write_all(format!("{}", CURRENT_VERSION).as_bytes()).map_err(db_err)?;
+pub fn update_version(path: &Path) -> io::Result<()> {
+	fs::create_dir_all(path)?;
+	let mut file = fs::File::create(version_file_path(path))?;
+	file.write_all(format!("{}", CURRENT_VERSION).as_bytes())?;
 	Ok(())
 }
 
@@ -158,7 +170,7 @@ fn version_file_path(path: &Path) -> PathBuf {
 mod tests {
 	use super::*;
 	use crate::{
-		tests::Block, DatabaseSettings, DatabaseSettingsSrc, KeepBlocks, TransactionStorageMode,
+		tests::Block, DatabaseSettings, DatabaseSource, KeepBlocks, TransactionStorageMode,
 	};
 	use sc_state_db::PruningMode;
 
@@ -176,7 +188,7 @@ mod tests {
 				state_cache_size: 0,
 				state_cache_child_ratio: None,
 				state_pruning: PruningMode::ArchiveAll,
-				source: DatabaseSettingsSrc::RocksDb { path: db_path.to_owned(), cache_size: 128 },
+				source: DatabaseSource::RocksDb { path: db_path.to_owned(), cache_size: 128 },
 				keep_blocks: KeepBlocks::All,
 				transaction_storage: TransactionStorageMode::BlockBody,
 			},

substrate/client/db/src/utils.rs

@@ -19,11 +19,11 @@
 //! Db-based backend utility structures and functions, used by both
 //! full and light storages.
 
-use std::{convert::TryInto, sync::Arc};
+use std::{convert::TryInto, fmt, io, path::Path, sync::Arc};
 
 use log::debug;
 
-use crate::{Database, DatabaseSettings, DatabaseSettingsSrc, DbHash};
+use crate::{Database, DatabaseSettings, DatabaseSource, DbHash};
 use codec::Decode;
 use sp_database::Transaction;
 use sp_runtime::{
@@ -204,88 +204,170 @@ where
 	})
 }
 
+fn backend_err(feat: &'static str) -> sp_blockchain::Error {
+	sp_blockchain::Error::Backend(feat.to_string())
+}
+
 /// Opens the configured database.
 pub fn open_database<Block: BlockT>(
 	config: &DatabaseSettings,
 	db_type: DatabaseType,
 ) -> sp_blockchain::Result<Arc<dyn Database<DbHash>>> {
-	#[allow(unused)]
-	fn db_open_error(feat: &'static str) -> sp_blockchain::Error {
-		sp_blockchain::Error::Backend(format!(
-			"`{}` feature not enabled, database can not be opened",
-			feat
-		))
-	}
-
 	let db: Arc<dyn Database<DbHash>> = match &config.source {
-		#[cfg(any(feature = "with-kvdb-rocksdb", test))]
-		DatabaseSettingsSrc::RocksDb { path, cache_size } => {
-			// first upgrade database to required version
-			crate::upgrade::upgrade_db::<Block>(&path, db_type)?;
-
-			// and now open database assuming that it has the latest version
-			let mut db_config = kvdb_rocksdb::DatabaseConfig::with_columns(NUM_COLUMNS);
-			let path = path
-				.to_str()
-				.ok_or_else(|| sp_blockchain::Error::Backend("Invalid database path".into()))?;
-
-			let mut memory_budget = std::collections::HashMap::new();
-			match db_type {
-				DatabaseType::Full => {
-					let state_col_budget = (*cache_size as f64 * 0.9) as usize;
-					let other_col_budget =
-						(cache_size - state_col_budget) / (NUM_COLUMNS as usize - 1);
-
-					for i in 0..NUM_COLUMNS {
-						if i == crate::columns::STATE {
-							memory_budget.insert(i, state_col_budget);
-						} else {
-							memory_budget.insert(i, other_col_budget);
-						}
-					}
-					log::trace!(
-						target: "db",
-						"Open RocksDB database at {}, state column budget: {} MiB, others({}) column cache: {} MiB",
-						path,
-						state_col_budget,
-						NUM_COLUMNS,
-						other_col_budget,
-					);
-				},
-				DatabaseType::Light => {
-					let col_budget = cache_size / (NUM_COLUMNS as usize);
-					for i in 0..NUM_COLUMNS {
-						memory_budget.insert(i, col_budget);
-					}
-					log::trace!(
-						target: "db",
-						"Open RocksDB light database at {}, column cache: {} MiB",
-						path,
-						col_budget,
-					);
-				},
+		DatabaseSource::ParityDb { path } => open_parity_db::<Block>(&path, db_type, true)?,
+		DatabaseSource::RocksDb { path, cache_size } =>
+			open_kvdb_rocksdb::<Block>(&path, db_type, true, *cache_size)?,
+		DatabaseSource::Custom(db) => db.clone(),
+		DatabaseSource::Auto { paritydb_path, rocksdb_path, cache_size } => {
+			// check if rocksdb exists first, if not, open paritydb
+			match open_kvdb_rocksdb::<Block>(&rocksdb_path, db_type, false, *cache_size) {
+				Ok(db) => db,
+				Err(OpenDbError::NotEnabled(_)) | Err(OpenDbError::DoesNotExist) =>
+					open_parity_db::<Block>(&paritydb_path, db_type, true)?,
+				Err(_) => return Err(backend_err("cannot open rocksdb. corrupted database")),
 			}
-			db_config.memory_budget = memory_budget;
-
-			let db = kvdb_rocksdb::Database::open(&db_config, &path)
-				.map_err(|err| sp_blockchain::Error::Backend(format!("{}", err)))?;
-			sp_database::as_database(db)
 		},
-		#[cfg(not(any(feature = "with-kvdb-rocksdb", test)))]
-		DatabaseSettingsSrc::RocksDb { .. } => return Err(db_open_error("with-kvdb-rocksdb")),
-		#[cfg(feature = "with-parity-db")]
-		DatabaseSettingsSrc::ParityDb { path } => crate::parity_db::open(&path, db_type)
-			.map_err(|e| sp_blockchain::Error::Backend(format!("{}", e)))?,
-		#[cfg(not(feature = "with-parity-db"))]
-		DatabaseSettingsSrc::ParityDb { .. } => return Err(db_open_error("with-parity-db")),
-		DatabaseSettingsSrc::Custom(db) => db.clone(),
 	};
 
 	check_database_type(&*db, db_type)?;
-
 	Ok(db)
 }
 
+#[derive(Debug)]
+enum OpenDbError {
+	// constructed only when rocksdb and paritydb are disabled
+	#[allow(dead_code)]
+	NotEnabled(&'static str),
+	DoesNotExist,
+	Internal(String),
+}
+
+type OpenDbResult = Result<Arc<dyn Database<DbHash>>, OpenDbError>;
+
+impl fmt::Display for OpenDbError {
+	fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+		match self {
+			OpenDbError::Internal(e) => write!(f, "{}", e.to_string()),
+			OpenDbError::DoesNotExist => write!(f, "Database does not exist at given location"),
+			OpenDbError::NotEnabled(feat) =>
+				write!(f, "`{}` feature not enabled, database can not be opened", feat),
+		}
+	}
+}
+
+impl From<OpenDbError> for sp_blockchain::Error {
+	fn from(err: OpenDbError) -> Self {
+		sp_blockchain::Error::Backend(err.to_string())
+	}
+}
+
+#[cfg(feature = "with-parity-db")]
+impl From<parity_db::Error> for OpenDbError {
+	fn from(err: parity_db::Error) -> Self {
+		if err.to_string().contains("use open_or_create") {
+			OpenDbError::DoesNotExist
+		} else {
+			OpenDbError::Internal(err.to_string())
+		}
+	}
+}
+
+impl From<io::Error> for OpenDbError {
+	fn from(err: io::Error) -> Self {
+		if err.to_string().contains("create_if_missing is false") {
+			OpenDbError::DoesNotExist
+		} else {
+			OpenDbError::Internal(err.to_string())
+		}
+	}
+}
+
+#[cfg(feature = "with-parity-db")]
+fn open_parity_db<Block: BlockT>(path: &Path, db_type: DatabaseType, create: bool) -> OpenDbResult {
+	let db = crate::parity_db::open(path, db_type, create)?;
+	Ok(db)
+}
+
+#[cfg(not(feature = "with-parity-db"))]
+fn open_parity_db<Block: BlockT>(
+	_path: &Path,
+	_db_type: DatabaseType,
+	_create: bool,
+) -> OpenDbResult {
+	Err(OpenDbError::NotEnabled("with-parity-db"))
+}
+
+#[cfg(any(feature = "with-kvdb-rocksdb", test))]
+fn open_kvdb_rocksdb<Block: BlockT>(
+	path: &Path,
+	db_type: DatabaseType,
+	create: bool,
+	cache_size: usize,
+) -> OpenDbResult {
+	// first upgrade database to required version
+	match crate::upgrade::upgrade_db::<Block>(&path, db_type) {
+		// in case of missing version file, assume that database simply does not exist at given location
+		Ok(_) | Err(crate::upgrade::UpgradeError::MissingDatabaseVersionFile) => (),
+		Err(err) => return Err(io::Error::new(io::ErrorKind::Other, err.to_string()).into()),
+	}
+
+	// and now open database assuming that it has the latest version
+	let mut db_config = kvdb_rocksdb::DatabaseConfig::with_columns(NUM_COLUMNS);
+	db_config.create_if_missing = create;
+
+	let mut memory_budget = std::collections::HashMap::new();
+	match db_type {
+		DatabaseType::Full => {
+			let state_col_budget = (cache_size as f64 * 0.9) as usize;
+			let other_col_budget = (cache_size - state_col_budget) / (NUM_COLUMNS as usize - 1);
+
+			for i in 0..NUM_COLUMNS {
+				if i == crate::columns::STATE {
+					memory_budget.insert(i, state_col_budget);
+				} else {
+					memory_budget.insert(i, other_col_budget);
+				}
+			}
+			log::trace!(
+				target: "db",
+				"Open RocksDB database at {:?}, state column budget: {} MiB, others({}) column cache: {} MiB",
+				path,
+				state_col_budget,
+				NUM_COLUMNS,
+				other_col_budget,
+			);
+		},
+		DatabaseType::Light => {
+			let col_budget = cache_size / (NUM_COLUMNS as usize);
+			for i in 0..NUM_COLUMNS {
+				memory_budget.insert(i, col_budget);
+			}
+			log::trace!(
+				target: "db",
+				"Open RocksDB light database at {:?}, column cache: {} MiB",
+				path,
+				col_budget,
+			);
+		},
+	}
+	db_config.memory_budget = memory_budget;
+
+	let db = kvdb_rocksdb::Database::open(&db_config, path)?;
+	// write database version only after the database is succesfully opened
+	crate::upgrade::update_version(path)?;
+	Ok(sp_database::as_database(db))
+}
+
+#[cfg(not(any(feature = "with-kvdb-rocksdb", test)))]
+fn open_kvdb_rocksdb<Block: BlockT>(
+	_path: &Path,
+	_db_type: DatabaseType,
+	_create: bool,
+	_cache_size: usize,
+) -> OpenDbResult {
+	Err(OpenDbError::NotEnabled("with-kvdb-rocksdb"))
+}
+
 /// Check database type.
 pub fn check_database_type(
 	db: &dyn Database<DbHash>,
@@ -482,7 +564,9 @@ impl<'a, 'b> codec::Input for JoinInput<'a, 'b> {
 #[cfg(test)]
 mod tests {
 	use super::*;
+	use crate::{KeepBlocks, TransactionStorageMode};
 	use codec::Input;
+	use sc_state_db::PruningMode;
 	use sp_runtime::testing::{Block as RawBlock, ExtrinsicWrapper};
 	type Block = RawBlock<ExtrinsicWrapper<u32>>;
 
@@ -521,4 +605,141 @@ mod tests {
 		assert_eq!(test, [7, 8, 6]);
 		assert_eq!(joined.remaining_len().unwrap(), Some(0));
 	}
+
+	fn db_settings(source: DatabaseSource) -> DatabaseSettings {
+		DatabaseSettings {
+			state_cache_size: 0,
+			state_cache_child_ratio: None,
+			state_pruning: PruningMode::ArchiveAll,
+			source,
+			keep_blocks: KeepBlocks::All,
+			transaction_storage: TransactionStorageMode::BlockBody,
+		}
+	}
+
+	#[cfg(feature = "with-parity-db")]
+	#[cfg(any(feature = "with-kvdb-rocksdb", test))]
+	#[test]
+	fn test_open_database_auto_new() {
+		let db_dir = tempfile::TempDir::new().unwrap();
+		let db_path = db_dir.path().to_owned();
+		let paritydb_path = db_path.join("paritydb");
+		let rocksdb_path = db_path.join("rocksdb_path");
+		let source = DatabaseSource::Auto {
+			paritydb_path: paritydb_path.clone(),
+			rocksdb_path: rocksdb_path.clone(),
+			cache_size: 128,
+		};
+		let mut settings = db_settings(source);
+
+		// it should create new auto (paritydb) database
+		{
+			let db_res = open_database::<Block>(&settings, DatabaseType::Full);
+			assert!(db_res.is_ok(), "New database should be created.");
+		}
+
+		// it should reopen existing auto (pairtydb) database
+		{
+			let db_res = open_database::<Block>(&settings, DatabaseType::Full);
+			assert!(db_res.is_ok(), "Existing parity database should be reopened");
+		}
+
+		// it should fail to open existing auto (pairtydb) database
+		{
+			settings.source = DatabaseSource::RocksDb { path: rocksdb_path, cache_size: 128 };
+			let db_res = open_database::<Block>(&settings, DatabaseType::Full);
+			assert!(db_res.is_ok(), "New database should be opened.");
+		}
+
+		// it should reopen existing auto (pairtydb) database
+		{
+			settings.source = DatabaseSource::ParityDb { path: paritydb_path };
+			let db_res = open_database::<Block>(&settings, DatabaseType::Full);
+			assert!(db_res.is_ok(), "Existing parity database should be reopened");
+		}
+	}
+
+	#[cfg(feature = "with-parity-db")]
+	#[cfg(any(feature = "with-kvdb-rocksdb", test))]
+	#[test]
+	fn test_open_database_rocksdb_new() {
+		let db_dir = tempfile::TempDir::new().unwrap();
+		let db_path = db_dir.path().to_owned();
+		let paritydb_path = db_path.join("paritydb");
+		let rocksdb_path = db_path.join("rocksdb_path");
+
+		let source = DatabaseSource::RocksDb { path: rocksdb_path.clone(), cache_size: 128 };
+		let mut settings = db_settings(source);
+
+		// it should create new rocksdb database
+		{
+			let db_res = open_database::<Block>(&settings, DatabaseType::Full);
+			assert!(db_res.is_ok(), "New rocksdb database should be created");
+		}
+
+		// it should reopen existing auto (rocksdb) database
+		{
+			settings.source = DatabaseSource::Auto {
+				paritydb_path: paritydb_path.clone(),
+				rocksdb_path: rocksdb_path.clone(),
+				cache_size: 128,
+			};
+			let db_res = open_database::<Block>(&settings, DatabaseType::Full);
+			assert!(db_res.is_ok(), "Existing rocksdb database should be reopened");
+		}
+
+		// it should fail to open existing auto (rocksdb) database
+		{
+			settings.source = DatabaseSource::ParityDb { path: paritydb_path };
+			let db_res = open_database::<Block>(&settings, DatabaseType::Full);
+			assert!(db_res.is_ok(), "New paritydb database should be created");
+		}
+
+		// it should reopen existing auto (pairtydb) database
+		{
+			settings.source = DatabaseSource::RocksDb { path: rocksdb_path, cache_size: 128 };
+			let db_res = open_database::<Block>(&settings, DatabaseType::Full);
+			assert!(db_res.is_ok(), "Existing rocksdb database should be reopened");
+		}
+	}
+
+	#[cfg(feature = "with-parity-db")]
+	#[cfg(any(feature = "with-kvdb-rocksdb", test))]
+	#[test]
+	fn test_open_database_paritydb_new() {
+		let db_dir = tempfile::TempDir::new().unwrap();
+		let db_path = db_dir.path().to_owned();
+		let paritydb_path = db_path.join("paritydb");
+		let rocksdb_path = db_path.join("rocksdb_path");
+
+		let source = DatabaseSource::ParityDb { path: paritydb_path.clone() };
+		let mut settings = db_settings(source);
+
+		// it should create new paritydb database
+		{
+			let db_res = open_database::<Block>(&settings, DatabaseType::Full);
+			assert!(db_res.is_ok(), "New database should be created.");
+		}
+
+		// it should reopen existing pairtydb database
+		{
+			let db_res = open_database::<Block>(&settings, DatabaseType::Full);
+			assert!(db_res.is_ok(), "Existing parity database should be reopened");
+		}
+
+		// it should fail to open existing pairtydb database
+		{
+			settings.source =
+				DatabaseSource::RocksDb { path: rocksdb_path.clone(), cache_size: 128 };
+			let db_res = open_database::<Block>(&settings, DatabaseType::Full);
+			assert!(db_res.is_ok(), "New rocksdb database should be created");
+		}
+
+		// it should reopen existing auto (pairtydb) database
+		{
+			settings.source = DatabaseSource::Auto { paritydb_path, rocksdb_path, cache_size: 128 };
+			let db_res = open_database::<Block>(&settings, DatabaseType::Full);
+			assert!(db_res.is_ok(), "Existing parity database should be reopened");
+		}
+	}
 }