Parity-db Change missing implementation.

client/db/src/parity_db.rs

@@ -86,15 +86,60 @@ 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(
+				format!(
+					"Ref counted operation on non ref counted columns {:?}",
+					not_ref_counted_column
+				)
+				.into(),
+			))
+		}
 
-		Ok(())
+		result.map_err(|e| DatabaseError(format!("Paritydb error: {:?}", e).into()))
 	}
 
 	fn get(&self, col: ColumnId, key: &[u8]) -> Option<Vec<u8>> {