Parity-db Change missing implementation. (#11049)

* support for release as in kvdb (only if no rc).

* Start impl

* minimal implementation for paritydb rc

* Update client/db/src/parity_db.rs

Co-authored-by: Arkadiy Paronyan <arkady.paronyan@gmail.com>

* Update client/db/src/parity_db.rs

Co-authored-by: Arkadiy Paronyan <arkady.paronyan@gmail.com>

* Update client/db/src/parity_db.rs

Co-authored-by: Arkadiy Paronyan <arkady.paronyan@gmail.com>

* Commit not panicking in DbAdapter

* errors from string

* update parity db version

Co-authored-by: Arkadiy Paronyan <arkady.paronyan@gmail.com>

substrate/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
+					},
+			})
+		}));
 
-		Ok(())
+		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
+			)))))
+		}
+
+		result.map_err(|e| DatabaseError(Box::new(e)))
 	}
 
 	fn get(&self, col: ColumnId, key: &[u8]) -> Option<Vec<u8>> {