Parity-db Change missing implementation.

client/db/Cargo.toml

@@ -22,7 +22,7 @@ kvdb-memorydb = "0.11.0"
 kvdb-rocksdb = { version = "0.15.2", optional = true }
 linked-hash-map = "0.5.4"
 log = "0.4.16"
-parity-db = { version = "0.3.12", optional = true }
+parity-db = { version = "0.3.13", optional = true }
 parking_lot = "0.12.0"
 sc-client-api = { version = "4.0.0-dev", path = "../api" }
 sc-state-db = { version = "0.10.0-dev", path = "../state-db" }

client/db/src/parity_db.rs

@@ -86,15 +86,57 @@ pub fn open<H: Clone + AsRef<[u8]>>(
 	Ok(std::sync::Arc::new(DbAdapter(db)))
 }
 
+fn ref_counted_column(col: u32) -> bool {
+	col == columns::TRANSACTION || col == columns::STATE
+}
+
 impl<H: Clone + AsRef<[u8]>> Database<H> for DbAdapter {
 	fn commit(&self, transaction: Transaction<H>) -> Result<(), DatabaseError> {
-		handle_err(self.0.commit(transaction.0.into_iter().map(|change| match change {
-			Change::Set(col, key, value) => (col as u8, key, Some(value)),
-			Change::Remove(col, key) => (col as u8, key, None),
-			_ => unimplemented!(),
-		})));
+		let mut not_ref_counted_column = Vec::new();
+		let result = self.0.commit(transaction.0.into_iter().filter_map(|change| {
+			Some(match change {
+				Change::Set(col, key, value) => (col as u8, key, Some(value)),
+				Change::Remove(col, key) => (col as u8, key, None),
+				Change::Store(col, key, value) =>
+					if ref_counted_column(col) {
+						(col as u8, key.as_ref().to_vec(), Some(value))
+					} else {
+						if !not_ref_counted_column.contains(&col) {
+							not_ref_counted_column.push(col);
+						}
+						return None
+					},
+				Change::Reference(col, key) =>
+					if ref_counted_column(col) {
+						// FIXME accessing value is not strictly needed, optimize this in parity-db.
+						let value = <Self as Database<H>>::get(self, col, key.as_ref());
+						(col as u8, key.as_ref().to_vec(), value)
+					} else {
+						if !not_ref_counted_column.contains(&col) {
+							not_ref_counted_column.push(col);
+						}
+						return None
+					},
+				Change::Release(col, key) =>
+					if ref_counted_column(col) {
+						(col as u8, key.as_ref().to_vec(), None)
+					} else {
+						if !not_ref_counted_column.contains(&col) {
+							not_ref_counted_column.push(col);
+						}
+						return None
+					},
+			})
+		}));
+
+		if not_ref_counted_column.len() > 0 {
+			return Err(DatabaseError(Box::new(parity_db::Error::InvalidInput(format!(
+				"Ref counted operation on non ref counted columns {:?}",
+				not_ref_counted_column
+			)))))
+		}
 
-		Ok(())
+		result.map_err(|e| DatabaseError(Box::new(e)))
 	}
 
 	fn get(&self, col: ColumnId, key: &[u8]) -> Option<Vec<u8>> {