Delete by prefix operator in kvdb

kvdb-memorydb/Cargo.toml

@@ -13,4 +13,4 @@ parking_lot = "0.10.0"
 kvdb = { version = "0.5", path = "../kvdb" }
 
 [dev-dependencies]
-kvdb-shared-tests = { path = "../kvdb-shared-tests", version = "0.2" }
+kvdb-shared-tests = { path = "../kvdb-shared-tests", version = "0.3" }

kvdb-memorydb/src/lib.rs

@@ -66,7 +66,24 @@ impl KeyValueDB for InMemory {
 					if let Some(col) = columns.get_mut(&col) {
 						col.remove(&*key);
 					}
-				}
+				},
+				DBOp::DeletePrefix { col, prefix } => {
+					if let Some(col) = columns.get_mut(&col) {
+						use std::ops::Bound;
+						if prefix.len() == 0 {
+							col.clear();
+						} else {
+							let start_range = Bound::Included(prefix.to_vec());
+							let end_range = Bound::Excluded(kvdb::end_prefix(&prefix[..]));
+							let keys: Vec<_> = col.range((start_range, end_range))
+								.map(|(k, _)| k.clone())
+								.collect();
+							for key in keys.into_iter() {
+								col.remove(&key[..]);
+							}
+						}
+					}
+				},
 			}
 		}
 		Ok(())
@@ -112,36 +129,49 @@ mod tests {
 	#[test]
 	fn get_fails_with_non_existing_column() -> io::Result<()> {
 		let db = create(1);
-		st::test_get_fails_with_non_existing_column(&db)
+		st::test_get_fails_with_non_existing_column(&db)?;
+		Ok(())
 	}
 
 	#[test]
 	fn put_and_get() -> io::Result<()> {
 		let db = create(1);
-		st::test_put_and_get(&db)
+		st::test_put_and_get(&db)?;
+		Ok(())
 	}
 
 	#[test]
 	fn delete_and_get() -> io::Result<()> {
 		let db = create(1);
-		st::test_delete_and_get(&db)
+		st::test_delete_and_get(&db)?;
+		Ok(())
+	}
+
+	#[test]
+	fn delete_prefix() -> io::Result<()> {
+		let db = create(st::NB_DELETE_PREFIX_TESTS);
+		st::test_delete_prefix(&db)?;
+		Ok(())
 	}
 
 	#[test]
 	fn iter() -> io::Result<()> {
 		let db = create(1);
-		st::test_iter(&db)
+		st::test_iter(&db)?;
+		Ok(())
 	}
 
 	#[test]
 	fn iter_from_prefix() -> io::Result<()> {
 		let db = create(1);
-		st::test_iter_from_prefix(&db)
+		st::test_iter_from_prefix(&db)?;
+		Ok(())
 	}
 
 	#[test]
 	fn complex() -> io::Result<()> {
 		let db = create(1);
-		st::test_complex(&db)
+		st::test_complex(&db)?;
+		Ok(())
 	}
 }

kvdb-rocksdb/Cargo.toml

@@ -27,7 +27,7 @@ parity-util-mem = { path = "../parity-util-mem", version = "0.6", default-featur
 alloc_counter = "0.0.4"
 criterion = "0.3"
 ethereum-types = { path = "../ethereum-types" }
-kvdb-shared-tests = { path = "../kvdb-shared-tests", version = "0.2" }
+kvdb-shared-tests = { path = "../kvdb-shared-tests", version = "0.3" }
 rand = "0.7.2"
 tempdir = "0.3.7"
 keccak-hash = { path = "../keccak-hash" }

kvdb-rocksdb/src/lib.rs

@@ -443,6 +443,18 @@ impl Database {
 							stats_total_bytes += key.len();
 							batch.delete_cf(cf, &key).map_err(other_io_err)?
 						}
+						DBOp::DeletePrefix { col: _, prefix } => {
+							if prefix.len() > 0 {
+								let end_range = kvdb::end_prefix(&prefix[..]);
+								batch.delete_range_cf(cf, &prefix[..], &end_range[..]).map_err(other_io_err)?;
+							} else {
+								// delete the whole column
+								let end_range = &[255u8];
+								batch.delete_range_cf(cf, &prefix[..], &end_range[..]).map_err(other_io_err)?;
+								batch.delete_cf(cf, &end_range[..]).map_err(other_io_err)?;
+								batch.delete_cf(cf, &[]).map_err(other_io_err)?;
+							}
+						}
 					};
 				}
 				self.stats.tally_bytes_written(stats_total_bytes as u64);
@@ -690,43 +702,57 @@ mod tests {
 	#[test]
 	fn get_fails_with_non_existing_column() -> io::Result<()> {
 		let db = create(1)?;
-		st::test_get_fails_with_non_existing_column(&db)
+		st::test_get_fails_with_non_existing_column(&db)?;
+		Ok(())
 	}
 
 	#[test]
 	fn put_and_get() -> io::Result<()> {
 		let db = create(1)?;
-		st::test_put_and_get(&db)
+		st::test_put_and_get(&db)?;
+		Ok(())
 	}
 
 	#[test]
 	fn delete_and_get() -> io::Result<()> {
 		let db = create(1)?;
-		st::test_delete_and_get(&db)
+		st::test_delete_and_get(&db)?;
+		Ok(())
+	}
+
+	#[test]
+	fn delete_prefix() -> io::Result<()> {
+		let db = create(st::NB_DELETE_PREFIX_TESTS)?;
+		st::test_delete_prefix(&db)?;
+		Ok(())
 	}
 
 	#[test]
 	fn iter() -> io::Result<()> {
 		let db = create(1)?;
-		st::test_iter(&db)
+		st::test_iter(&db)?;
+		Ok(())
 	}
 
 	#[test]
 	fn iter_from_prefix() -> io::Result<()> {
 		let db = create(1)?;
-		st::test_iter_from_prefix(&db)
+		st::test_iter_from_prefix(&db)?;
+		Ok(())
 	}
 
 	#[test]
 	fn complex() -> io::Result<()> {
 		let db = create(1)?;
-		st::test_complex(&db)
+		st::test_complex(&db)?;
+		Ok(())
 	}
 
 	#[test]
 	fn stats() -> io::Result<()> {
 		let db = create(3)?;
-		st::test_io_stats(&db)
+		st::test_io_stats(&db)?;
+		Ok(())
 	}
 
 	#[test]

kvdb-shared-tests/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "kvdb-shared-tests"
-version = "0.2.0"
+version = "0.3.0"
 authors = ["Parity Technologies <[email protected]>"]
 edition = "2018"
 description = "Shared tests for kvdb functionality, to be executed against actual implementations"

kvdb-shared-tests/src/lib.rs

@@ -171,6 +171,62 @@ pub fn test_io_stats(db: &dyn KeyValueDB) -> io::Result<()> {
 	Ok(())
 }
 
+pub const NB_DELETE_PREFIX_TESTS: u32 = 5;
+
+/// A test for `KeyValueDB::delete_prefix`.
+pub fn test_delete_prefix(db: &dyn KeyValueDB) -> io::Result<()> {
+	let keys = [
+		&[][..],
+		&[0u8][..],
+		&[0, 1][..],
+		&[1][..],
+		&[1, 0][..],
+		&[1, 255][..],
+		&[1, 255, 255][..],
+		&[2][..],
+		&[2, 0][..],
+		&[2, 255][..],
+	];
+	let init_db = |ix: u32| -> io::Result<()> {
+		let mut batch = db.transaction();
+		for (i, key) in keys.iter().enumerate() {
+			batch.put(ix, key, &[i as u8]);
+		}
+		db.write(batch)?;
+		Ok(())
+	};
+	let check_db = |ix: u32, content: [bool; 10]| -> io::Result<()> {
+		let mut state = [true; 10];
+		for (c, key) in keys.iter().enumerate() {
+			state[c] = db.get(ix, key)?.is_some();
+		}
+		assert_eq!(state, content, "at {}", ix);
+		Ok(())
+	};
+	let tests: [_; NB_DELETE_PREFIX_TESTS as usize] = [
+		// standard
+		(&[1u8][..], [true, true, true, false, false, false, false, true, true, true]),
+		// edge
+		(&[1u8, 255, 255][..], [true, true, true, true, true, true, false, true, true, true]),
+		// none 1
+		(&[1, 2][..], [true, true, true, true, true, true, true, true, true, true]),
+		// none 2
+		(&[8][..], [true, true, true, true, true, true, true, true, true, true]),
+		// all
+		(&[][..], [false, false, false, false, false, false, false, false, false, false]),
+	];
+	for (ix, test) in tests.iter().enumerate() {
+		let ix = ix as u32;
+		init_db(ix)?;
+		let mut batch = db.transaction();
+		batch.delete_prefix(ix, test.0);
+		db.write(batch)?;
+		check_db(ix, test.1)?;
+	}
+
+	Ok(())
+}
+
 /// A complex test.
 pub fn test_complex(db: &dyn KeyValueDB) -> io::Result<()> {
 	let key1 = b"02c69be41d0b7e40352fc85be1cd65eb03d40ef8427a0ca4596b1ead9a00e9fc";

kvdb-web/Cargo.toml

@@ -34,11 +34,12 @@ features = [
 	'EventTarget',
 	'IdbCursor',
 	'IdbCursorWithValue',
+	'IdbKeyRange',
 	'DomStringList',
 ]
 
 [dev-dependencies]
 console_log = "0.1.2"
-kvdb-shared-tests = { path = "../kvdb-shared-tests", version = "0.2" }
+kvdb-shared-tests = { path = "../kvdb-shared-tests", version = "0.3" }
 wasm-bindgen-test = "0.3.4"
 wasm-bindgen-futures = "0.4.4"

kvdb-web/src/indexed_db.rs

@@ -10,7 +10,7 @@
 
 use js_sys::{Array, ArrayBuffer, Uint8Array};
 use wasm_bindgen::{closure::Closure, JsCast, JsValue};
-use web_sys::{Event, IdbCursorWithValue, IdbDatabase, IdbOpenDbRequest, IdbRequest, IdbTransactionMode};
+use web_sys::{Event, IdbCursorWithValue, IdbDatabase, IdbOpenDbRequest, IdbRequest, IdbTransactionMode, IdbKeyRange};
 
 use futures::channel;
 use futures::prelude::*;
@@ -157,6 +157,19 @@ pub fn idb_commit_transaction(idb: &IdbDatabase, txn: &DBTransaction, columns: u
 					warn!("error deleting key from col_{}: {:?}", column, err);
 				}
 			}
+			DBOp::DeletePrefix { col, prefix } => {
+				let column = *col as usize;
+				// Convert rust bytes to js arrays
+				let prefix_js_start = Uint8Array::from(prefix.as_ref());
+				let prefix_js_end = Uint8Array::from(prefix.as_ref());
+
+				let range = IdbKeyRange::bound(prefix_js_start.as_ref(), prefix_js_end.as_ref())
+					.expect("Starting and ending at same value is valid bound; qed");
+				let res = object_stores[column].delete(range.as_ref());
+				if let Err(err) = res {
+					warn!("error deleting prefix from col_{}: {:?}", column, err);
+				}
+			}
 		}
 	}
 

kvdb-web/tests/indexed_db.rs

@@ -39,6 +39,12 @@ async fn delete_and_get() {
 	st::test_delete_and_get(&db).unwrap()
 }
 
+#[wasm_bindgen_test]
+async fn delete_prefix() {
+	let db = open_db(st::NB_DELETE_PREFIX_TESTS, "delete_prefix").await;
+	st::test_delete_prefix(&db).unwrap()
+}
+
 #[wasm_bindgen_test]
 async fn iter() {
 	let db = open_db(1, "iter").await;

kvdb/src/lib.rs

@@ -35,6 +35,7 @@ pub struct DBTransaction {
 pub enum DBOp {
 	Insert { col: u32, key: DBKey, value: DBValue },
 	Delete { col: u32, key: DBKey },
+	DeletePrefix { col: u32, prefix: DBKey },
 }
 
 impl DBOp {
@@ -43,6 +44,7 @@ impl DBOp {
 		match *self {
 			DBOp::Insert { ref key, .. } => key,
 			DBOp::Delete { ref key, .. } => key,
+			DBOp::DeletePrefix { ref prefix, .. } => prefix,
 		}
 	}
 
@@ -51,6 +53,7 @@ impl DBOp {
 		match *self {
 			DBOp::Insert { col, .. } => col,
 			DBOp::Delete { col, .. } => col,
+			DBOp::DeletePrefix { col, .. } => col,
 		}
 	}
 }
@@ -80,6 +83,12 @@ impl DBTransaction {
 	pub fn delete(&mut self, col: u32, key: &[u8]) {
 		self.ops.push(DBOp::Delete { col, key: DBKey::from_slice(key) });
 	}
+
+	/// Delete value by key.
+	pub fn delete_prefix(&mut self, col: u32, prefix: &[u8]) {
+		self.ops.push(DBOp::DeletePrefix { col, prefix: DBKey::from_slice(prefix) });
+	}
+
 }
 
 /// Generic key-value database.
@@ -106,6 +115,17 @@ pub trait KeyValueDB: Sync + Send + parity_util_mem::MallocSizeOf {
 	/// Write a transaction of changes to the backing store.
 	fn write(&self, transaction: DBTransaction) -> io::Result<()>;
 
+	/// Delete all prefixed key (do not delete buffered values and
+	/// do not support atomic operation in a transaction).
+	fn delete_prefix(&self, col: u32, prefix: &[u8]) -> io::Result<()> {
+		// default unefficient implementation.
+		let mut transaction = DBTransaction::new();
+		self.iter_from_prefix(col, prefix).for_each(|(key, _)| {
+			transaction.delete_prefix(col, &key[..]);
+		});
+		self.write(transaction)
+	}
+
 	/// Iterate over the data for a given column.
 	fn iter<'a>(&'a self, col: u32) -> Box<dyn Iterator<Item = (Box<[u8]>, Box<[u8]>)> + 'a>;
 
@@ -129,3 +149,20 @@ pub trait KeyValueDB: Sync + Send + parity_util_mem::MallocSizeOf {
 		IoStats::empty()
 	}
 }
+
+/// Return for a start inclusive prefix, the non inclusive end.
+pub fn end_prefix(prefix: &[u8]) -> Vec<u8> {
+	let mut end_range = prefix.to_vec();
+	loop {
+		if end_range.len() == 0 {
+			return end_range;
+		}
+		let end_index = end_range.len() - 1;
+		if end_range[end_index] < u8::max_value() {
+			end_range[end_index] = end_range[end_index] + 1;
+			return end_range;
+		} else {
+			end_range.pop();
+		};
+	};
+}