Add add_column and test

Signed-off-by: Andrei Sandu <andrei-mihail@parity.io>

src/db.rs

@@ -25,9 +25,9 @@ use crate::{
 	hash::IdentityBuildHasher,
 	index::PlanOutcome,
 	log::{Log, LogAction},
-	options::Options,
+	options::{Options, CURRENT_VERSION},
 	stats::StatSummary,
-	Key,
+	ColumnOptions, Key,
 };
 use fs2::FileExt;
 use parking_lot::{Condvar, Mutex, RwLock};
@@ -1027,6 +1027,18 @@ impl Db {
 	pub fn stats(&self) -> StatSummary {
 		self.inner.stats()
 	}
+
+	/// Add a new column with options specified by `new_column_options`.
+	pub fn add_column(options: &mut Options, new_column_options: ColumnOptions) -> Result<()> {
+		options.columns.push(new_column_options);
+		options.write_metadata_with_version(
+			&options.path,
+			&options.salt.expect("Salt is required to add column"),
+			Some(CURRENT_VERSION),
+		)?;
+
+		Ok(())
+	}
 }
 
 impl Drop for Db {
@@ -1431,7 +1443,7 @@ impl EnableCommitPipelineStages {
 mod tests {
 	use crate::Value;
 
-	use super::{Db, EnableCommitPipelineStages, InternalOptions, Options};
+	use super::{ColumnOptions, Db, EnableCommitPipelineStages, InternalOptions, Options};
 	use rand::Rng;
 	use std::collections::{BTreeMap, HashSet};
 	use tempfile::tempdir;
@@ -1590,6 +1602,87 @@ mod tests {
 		assert_eq!(db.get(col_nb, key3.as_slice()).unwrap(), None);
 	}
 
+	#[test]
+	fn test_add_column() {
+		let tmp = tempdir().unwrap();
+		let mut options = Options::with_columns(tmp.path(), 1);
+		options.salt = Some(options.salt.unwrap_or_default());
+
+		let old_col_id = 0;
+		let new_col_id = 1;
+		let new_col_indexed_id = 2;
+
+		let key1 = b"key1".to_vec();
+		let key2 = b"key2".to_vec();
+		let key3 = b"key3".to_vec();
+
+		let db_test = EnableCommitPipelineStages::DbFile;
+		let inner_options =
+			InternalOptions { create: true, commit_stages: db_test, ..Default::default() };
+		let db = Db::open_inner(&options, &inner_options).unwrap();
+
+		db.commit(vec![
+			(old_col_id, key1.clone(), Some(b"value1".to_vec())),
+			(old_col_id, key2.clone(), Some(b"value2".to_vec())),
+			(old_col_id, key3.clone(), Some(b"value3".to_vec())),
+		])
+		.unwrap();
+		db_test.run_stages(&db);
+
+		drop(db);
+
+		Db::add_column(&mut options, ColumnOptions { btree_index: false, ..Default::default() })
+			.unwrap();
+
+		Db::add_column(&mut options, ColumnOptions { btree_index: true, ..Default::default() })
+			.unwrap();
+
+		let inner_options =
+			InternalOptions { create: true, commit_stages: db_test, ..Default::default() };
+		let mut options = Options::with_columns(tmp.path(), 3);
+		options.columns[new_col_indexed_id as usize].btree_index = true;
+
+		let db_test = EnableCommitPipelineStages::DbFile;
+		let db = Db::open_inner(&options, &inner_options).unwrap();
+
+		// Expected number of columns
+		assert_eq!(db.num_columns(), 3);
+
+		let new_key1 = b"abcdef".to_vec();
+		let new_key2 = b"123456".to_vec();
+
+		// Write to new columns.
+		db.commit(vec![
+			(new_col_id, new_key1.clone(), Some(new_key1.to_vec())),
+			(new_col_id, new_key2.clone(), Some(new_key2.to_vec())),
+			(new_col_indexed_id, new_key1.clone(), Some(new_key1.to_vec())),
+			(new_col_indexed_id, new_key2.clone(), Some(new_key2.to_vec())),
+		])
+		.unwrap();
+		db_test.run_stages(&db);
+
+		drop(db);
+
+		// Reopen DB and fetch all keys we inserted.
+		let db = Db::open_inner(&options, &inner_options).unwrap();
+
+		assert_eq!(db.get(old_col_id, key1.as_slice()).unwrap(), Some(b"value1".to_vec()));
+		assert_eq!(db.get(old_col_id, key2.as_slice()).unwrap(), Some(b"value2".to_vec()));
+		assert_eq!(db.get(old_col_id, key3.as_slice()).unwrap(), Some(b"value3".to_vec()));
+
+		// Fetch from new columns
+		assert_eq!(db.get(new_col_id, new_key1.as_slice()).unwrap(), Some(new_key1.to_vec()));
+		assert_eq!(db.get(new_col_id, new_key2.as_slice()).unwrap(), Some(new_key2.to_vec()));
+		assert_eq!(
+			db.get(new_col_indexed_id, new_key1.as_slice()).unwrap(),
+			Some(new_key1.to_vec())
+		);
+		assert_eq!(
+			db.get(new_col_indexed_id, new_key2.as_slice()).unwrap(),
+			Some(new_key2.to_vec())
+		);
+	}
+
 	#[test]
 	fn test_indexed_btree_1() {
 		test_indexed_btree_inner(EnableCommitPipelineStages::CommitOverlay, false);