Inner hashing of value in state trie (runtime versioning). (#9732)

* starting * Updated from other branch. * setting flag * flag in storage struct * fix flagging to access and insert. * added todo to fix * also missing serialize meta to storage proof * extract meta. * Isolate old trie layout. * failing test that requires storing in meta when old hash scheme is used. * old hash compatibility * Db migrate. * runing tests with both states when interesting. * fix chain spec test with serde default. * export state (missing trie function). * Pending using new branch, lacking genericity on layout resolution. * extract and set global meta * Update to branch 4 * fix iterator with root flag (no longer insert node). * fix trie root hashing of root * complete basic backend. * Remove old_hash meta from proof that do not use inner_hashing. * fix trie test for empty (force layout on empty deltas). * Root update fix. * debug on meta * Use trie key iteration that do not include value in proofs. * switch default test ext to use inner hash. * small integration test, and fix tx cache mgmt in ext. test failing * Proof scenario at state-machine level. * trace for db upgrade * try different param * act more like iter_from. * Bigger batches. * Update trie dependency. * drafting codec changes and refact * before removing unused branch no value alt hashing. more work todo rename all flag var to alt_hash, and remove extrinsic replace by storage query at every storage_root call. * alt hashing only for branch with value. * fix trie tests * Hash of value include the encoded size. * removing fields(broken) * fix trie_stream to also include value length in inner hash. * triedbmut only using alt type if inner hashing. * trie_stream to also only use alt hashing type when actually alt hashing. * Refactor meta state, logic should work with change of trie treshold. * Remove NoMeta variant. * Remove state_hashed trigger specific functions. * pending switching to using threshold, new storage root api does not make much sense. * refactoring to use state from backend (not possible payload changes). * Applying from previous state * Remove default from storage, genesis need a special build. * rem empty space * Catch problem: when using triedb with default: we should not revert nodes: otherwhise thing as trie codec cannot decode-encode without changing state. * fix compilation * Right logic to avoid switch on reencode when default layout. * Clean up some todos * remove trie meta from root upstream * update upstream and fix benches. * split some long lines. * UPdate trie crate to work with new design. * Finish update to refactored upstream. * update to latest triedb changes. * Clean up. * fix executor test. * rust fmt from master. * rust format. * rustfmt * fix * start host function driven versioning * update state-machine part * still need access to state version from runtime * state hash in mem: wrong * direction likely correct, but passing call to code exec for genesis init seem awkward. * state version serialize in runtime, wrong approach, just initialize it with no threshold for core api < 4 seems more proper. * stateversion from runtime version (core api >= 4). * update trie, fix tests * unused import * clean some TODOs * Require RuntimeVersionOf for executor * use RuntimeVersionOf to resolve genesis state version. * update runtime version test * fix state-machine tests * TODO * Use runtime version from storage wasm with fast sync. * rustfmt * fmt * fix test * revert useless changes. * clean some unused changes * fmt * removing useless trait function. * remove remaining reference to state_hash * fix some imports * Follow chain state version management. * trie update, fix and constant threshold for trie layouts. * update deps * Update to latest trie pr changes. * fix benches * Verify proof requires right layout. * update trie_root * Update trie deps to latest * Update to latest trie versioning * Removing patch * update lock * extrinsic for sc-service-test using layout v0. * Adding RuntimeVersionOf to CallExecutor works. * fmt * error when resolving version and no wasm in storage. * use existing utils to instantiate runtime code. * Patch to delay runtime switch. * Revert "Patch to delay runtime switch." This reverts commit 67e55fee468f1a0cda853f5362b22e0d775786da. * useless closure * remove remaining state_hash variables. * Remove outdated comment * useless inner hash * fmt * fmt and opt-in feature to apply state change. * feature gate core version, use new test feature for node and test node * Use a 'State' api version instead of Core one. * fix merge of test function * use blake macro. * Fix state api (require declaring the api in runtime). * Opt out feature, fix macro for io to select a given version instead of latest. * run test nodes on new state. * fix * Apply review change (docs and error). * fmt * use explicit runtime_interface in doc test * fix ui test * fix doc test * fmt * use default for path and specname when resolving version. * small review related changes. * doc value size requirement. * rename old_state feature * Remove macro changes * feature rename * state version as host function parameter * remove flag for client api * fix tests * switch storage chain proof to V1 * host functions, pass by state version enum * use WrappedRuntimeCode * start * state_version in runtime version * rust fmt * Update storage proof of max size. * fix runtime version rpc test * right intent of convert from compat * fix doc test * fix doc test * split proof * decode without replay, and remove some reexports. * Decode with compatibility by default. * switch state_version to u8. And remove RuntimeVersionBasis. * test * use api when reading embedded version * fix decode with apis * extract core version instead * test fix * unused import * review changes. Co-authored-by: kianenigma <kian@parity.io>
2026-06-12 20:31:13 +00:00 · 2021-12-24 09:54:07 +01:00
parent ea30c739ea
commit 4c651637f2
78 changed files with 1814 additions and 782 deletions
@@ -23,7 +23,7 @@ use crate::{
 };
 use codec::Encode;
 use hash_db::Hasher;
-use sp_core::storage::{ChildInfo, TrackedStorageKey};
+use sp_core::storage::{ChildInfo, StateVersion, TrackedStorageKey};
 #[cfg(feature = "std")]
 use sp_core::traits::RuntimeCode;
 use sp_std::vec::Vec;
@@ -140,6 +140,7 @@ pub trait Backend<H: Hasher>: sp_std::fmt::Debug {
 	fn storage_root<'a>(
 		&self,
 		delta: impl Iterator<Item = (&'a [u8], Option<&'a [u8]>)>,
+		state_version: StateVersion,
 	) -> (H::Out, Self::Transaction)
 	where
 		H::Out: Ord;
@@ -151,6 +152,7 @@ pub trait Backend<H: Hasher>: sp_std::fmt::Debug {
 		&self,
 		child_info: &ChildInfo,
 		delta: impl Iterator<Item = (&'a [u8], Option<&'a [u8]>)>,
+		state_version: StateVersion,
 	) -> (H::Out, bool, Self::Transaction)
 	where
 		H::Out: Ord;
@@ -176,7 +178,6 @@ pub trait Backend<H: Hasher>: sp_std::fmt::Debug {
 	fn as_trie_backend(&self) -> Option<&TrieBackend<Self::TrieBackendStorage, H>> {
 		None
 	}
-
 	/// Calculate the storage root, with given delta over what is already stored
 	/// in the backend, and produce a "transaction" that can be used to commit.
 	/// Does include child storage updates.
@@ -186,6 +187,7 @@ pub trait Backend<H: Hasher>: sp_std::fmt::Debug {
 		child_deltas: impl Iterator<
 			Item = (&'a ChildInfo, impl Iterator<Item = (&'a [u8], Option<&'a [u8]>)>),
 		>,
+		state_version: StateVersion,
 	) -> (H::Out, Self::Transaction)
 	where
 		H::Out: Ord + Encode,
@@ -194,7 +196,8 @@ pub trait Backend<H: Hasher>: sp_std::fmt::Debug {
 		let mut child_roots: Vec<_> = Default::default();
 		// child first
 		for (child_info, child_delta) in child_deltas {
-			let (child_root, empty, child_txs) = self.child_storage_root(&child_info, child_delta);
+			let (child_root, empty, child_txs) =
+				self.child_storage_root(&child_info, child_delta, state_version);
 			let prefixed_storage_key = child_info.prefixed_storage_key();
 			txs.consolidate(child_txs);
 			if empty {
@@ -205,8 +208,9 @@ pub trait Backend<H: Hasher>: sp_std::fmt::Debug {
 		}
 		let (root, parent_txs) = self.storage_root(
 			delta
-				.map(|(k, v)| (k, v.as_ref().map(|v| &v[..])))
+				.map(|(k, v)| (&k[..], v.as_ref().map(|v| &v[..])))
 				.chain(child_roots.iter().map(|(k, v)| (&k[..], v.as_ref().map(|v| &v[..])))),
+			state_version,
 		);
 		txs.consolidate(parent_txs);
 		(root, txs)
@@ -286,7 +290,11 @@ impl Consolidate for Vec<(Option<ChildInfo>, StorageCollection)> {
 	}
 }

-impl<H: Hasher, KF: sp_trie::KeyFunction<H>> Consolidate for sp_trie::GenericMemoryDB<H, KF> {
+impl<H, KF> Consolidate for sp_trie::GenericMemoryDB<H, KF>
+where
+	H: Hasher,
+	KF: sp_trie::KeyFunction<H>,
+{
 	fn consolidate(&mut self, other: Self) {
 		sp_trie::GenericMemoryDB::consolidate(self, other)
 	}
@@ -23,13 +23,14 @@ use hash_db::Hasher;
 use log::warn;
 use sp_core::{
 	storage::{
-		well_known_keys::is_child_storage_key, ChildInfo, Storage, StorageChild, TrackedStorageKey,
+		well_known_keys::is_child_storage_key, ChildInfo, StateVersion, Storage, StorageChild,
+		TrackedStorageKey,
 	},
 	traits::Externalities,
 	Blake2Hasher,
 };
 use sp_externalities::{Extension, Extensions};
-use sp_trie::{empty_child_trie_root, trie_types::Layout, TrieConfiguration};
+use sp_trie::{empty_child_trie_root, LayoutV0, LayoutV1, TrieConfiguration};
 use std::{
 	any::{Any, TypeId},
 	collections::BTreeMap,
@@ -273,7 +274,7 @@ impl Externalities for BasicExternalities {
 		crate::ext::StorageAppend::new(current).append(value);
 	}

-	fn storage_root(&mut self) -> Vec<u8> {
+	fn storage_root(&mut self, state_version: StateVersion) -> Vec<u8> {
 		let mut top = self.inner.top.clone();
 		let prefixed_keys: Vec<_> = self
 			.inner
@@ -284,9 +285,9 @@ impl Externalities for BasicExternalities {
 		// Single child trie implementation currently allows using the same child
 		// empty root for all child trie. Using null storage key until multiple
 		// type of child trie support.
-		let empty_hash = empty_child_trie_root::<Layout<Blake2Hasher>>();
+		let empty_hash = empty_child_trie_root::<LayoutV1<Blake2Hasher>>();
 		for (prefixed_storage_key, child_info) in prefixed_keys {
-			let child_root = self.child_storage_root(&child_info);
+			let child_root = self.child_storage_root(&child_info, state_version);
 			if &empty_hash[..] == &child_root[..] {
 				top.remove(prefixed_storage_key.as_slice());
 			} else {
@@ -294,17 +295,26 @@ impl Externalities for BasicExternalities {
 			}
 		}

-		Layout::<Blake2Hasher>::trie_root(self.inner.top.clone()).as_ref().into()
+		match state_version {
+			StateVersion::V0 =>
+				LayoutV0::<Blake2Hasher>::trie_root(self.inner.top.clone()).as_ref().into(),
+			StateVersion::V1 =>
+				LayoutV1::<Blake2Hasher>::trie_root(self.inner.top.clone()).as_ref().into(),
+		}
 	}

-	fn child_storage_root(&mut self, child_info: &ChildInfo) -> Vec<u8> {
+	fn child_storage_root(
+		&mut self,
+		child_info: &ChildInfo,
+		state_version: StateVersion,
+	) -> Vec<u8> {
 		if let Some(child) = self.inner.children_default.get(child_info.storage_key()) {
 			let delta = child.data.iter().map(|(k, v)| (k.as_ref(), Some(v.as_ref())));
 			crate::in_memory_backend::new_in_mem::<Blake2Hasher>()
-				.child_storage_root(&child.child_info, delta)
+				.child_storage_root(&child.child_info, delta, state_version)
 				.0
 		} else {
-			empty_child_trie_root::<Layout<Blake2Hasher>>()
+			empty_child_trie_root::<LayoutV1<Blake2Hasher>>()
 		}
 		.encode()
 	}
@@ -389,7 +399,7 @@ mod tests {
 		const ROOT: [u8; 32] =
 			hex!("39245109cef3758c2eed2ccba8d9b370a917850af3824bc8348d505df2c298fa");

-		assert_eq!(&ext.storage_root()[..], &ROOT);
+		assert_eq!(&ext.storage_root(StateVersion::default())[..], &ROOT);
 	}

 	#[test]
@@ -24,9 +24,11 @@ use codec::{Decode, Encode, EncodeAppend};
 use hash_db::Hasher;
 #[cfg(feature = "std")]
 use sp_core::hexdisplay::HexDisplay;
-use sp_core::storage::{well_known_keys::is_child_storage_key, ChildInfo, TrackedStorageKey};
+use sp_core::storage::{
+	well_known_keys::is_child_storage_key, ChildInfo, StateVersion, TrackedStorageKey,
+};
 use sp_externalities::{Extension, ExtensionStore, Externalities};
-use sp_trie::{empty_child_trie_root, trie_types::Layout};
+use sp_trie::{empty_child_trie_root, LayoutV1};

 use crate::{log_error, trace, warn, StorageTransactionCache};
 use sp_std::{
@@ -505,7 +507,7 @@ where
 		StorageAppend::new(current_value).append(value);
 	}

-	fn storage_root(&mut self) -> Vec<u8> {
+	fn storage_root(&mut self, state_version: StateVersion) -> Vec<u8> {
 		let _guard = guard();
 		if let Some(ref root) = self.storage_transaction_cache.transaction_storage_root {
 			trace!(
@@ -518,7 +520,9 @@ where
 			return root.encode()
 		}

-		let root = self.overlay.storage_root(self.backend, self.storage_transaction_cache);
+		let root =
+			self.overlay
+				.storage_root(self.backend, self.storage_transaction_cache, state_version);
 		trace!(
 			target: "state",
 			method = "StorageRoot",
@@ -529,7 +533,11 @@ where
 		root.encode()
 	}

-	fn child_storage_root(&mut self, child_info: &ChildInfo) -> Vec<u8> {
+	fn child_storage_root(
+		&mut self,
+		child_info: &ChildInfo,
+		state_version: StateVersion,
+	) -> Vec<u8> {
 		let _guard = guard();
 		let storage_key = child_info.storage_key();
 		let prefixed_storage_key = child_info.prefixed_storage_key();
@@ -537,7 +545,8 @@ where
 			let root = self
 				.storage(prefixed_storage_key.as_slice())
 				.and_then(|k| Decode::decode(&mut &k[..]).ok())
-				.unwrap_or_else(|| empty_child_trie_root::<Layout<H>>());
+				// V1 is equivalent to V0 on empty root.
+				.unwrap_or_else(|| empty_child_trie_root::<LayoutV1<H>>());
 			trace!(
 				target: "state",
 				method = "ChildStorageRoot",
@@ -550,7 +559,7 @@ where
 		} else {
 			let root = if let Some((changes, info)) = self.overlay.child_changes(storage_key) {
 				let delta = changes.map(|(k, v)| (k.as_ref(), v.value().map(AsRef::as_ref)));
-				Some(self.backend.child_storage_root(info, delta))
+				Some(self.backend.child_storage_root(info, delta, state_version))
 			} else {
 				None
 			};
@@ -583,7 +592,8 @@ where
 				let root = self
 					.storage(prefixed_storage_key.as_slice())
 					.and_then(|k| Decode::decode(&mut &k[..]).ok())
-					.unwrap_or_else(|| empty_child_trie_root::<Layout<H>>());
+					// V1 is equivalent to V0 on empty root.
+					.unwrap_or_else(|| empty_child_trie_root::<LayoutV1<H>>());

 				trace!(
 					target: "state",
@@ -648,7 +658,12 @@ where
 			self.overlay.rollback_transaction().expect(BENCHMARKING_FN);
 		}
 		self.overlay
-			.drain_storage_changes(self.backend, Default::default(), self.storage_transaction_cache)
+			.drain_storage_changes(
+				self.backend,
+				Default::default(),
+				self.storage_transaction_cache,
+				Default::default(), // using any state
+			)
 			.expect(EXT_NOT_ALLOWED_TO_FAIL);
 		self.backend.wipe().expect(EXT_NOT_ALLOWED_TO_FAIL);
 		self.mark_dirty();
@@ -658,12 +673,19 @@ where
 	}

 	fn commit(&mut self) {
+		// Bench always use latest state.
+		let state_version = StateVersion::default();
 		for _ in 0..self.overlay.transaction_depth() {
 			self.overlay.commit_transaction().expect(BENCHMARKING_FN);
 		}
 		let changes = self
 			.overlay
-			.drain_storage_changes(self.backend, Default::default(), self.storage_transaction_cache)
+			.drain_storage_changes(
+				self.backend,
+				Default::default(),
+				self.storage_transaction_cache,
+				state_version,
+			)
 			.expect(EXT_NOT_ALLOWED_TO_FAIL);
 		self.backend
 			.commit(
@@ -878,15 +900,18 @@ mod tests {
 		let mut overlay = OverlayedChanges::default();
 		overlay.set_storage(vec![20], None);
 		overlay.set_storage(vec![30], Some(vec![31]));
-		let backend = Storage {
-			top: map![
-				vec![10] => vec![10],
-				vec![20] => vec![20],
-				vec![40] => vec![40]
-			],
-			children_default: map![],
-		}
-		.into();
+		let backend = (
+			Storage {
+				top: map![
+					vec![10] => vec![10],
+					vec![20] => vec![20],
+					vec![40] => vec![40]
+				],
+				children_default: map![],
+			},
+			StateVersion::default(),
+		)
+			.into();

 		let ext = TestExt::new(&mut overlay, &mut cache, &backend, None);

@@ -924,13 +949,16 @@ mod tests {
 		overlay.set_storage(vec![27], None);
 		overlay.set_storage(vec![28], None);
 		overlay.set_storage(vec![29], None);
-		let backend = Storage {
-			top: map![
-				vec![30] => vec![30]
-			],
-			children_default: map![],
-		}
-		.into();
+		let backend = (
+			Storage {
+				top: map![
+					vec![30] => vec![30]
+				],
+				children_default: map![],
+			},
+			StateVersion::default(),
+		)
+			.into();

 		let ext = TestExt::new(&mut overlay, &mut cache, &backend, None);

@@ -948,20 +976,23 @@ mod tests {
 		let mut overlay = OverlayedChanges::default();
 		overlay.set_child_storage(child_info, vec![20], None);
 		overlay.set_child_storage(child_info, vec![30], Some(vec![31]));
-		let backend = Storage {
-			top: map![],
-			children_default: map![
-				child_info.storage_key().to_vec() => StorageChild {
-					data: map![
-						vec![10] => vec![10],
-						vec![20] => vec![20],
-						vec![40] => vec![40]
-					],
-					child_info: child_info.to_owned(),
-				}
-			],
-		}
-		.into();
+		let backend = (
+			Storage {
+				top: map![],
+				children_default: map![
+					child_info.storage_key().to_vec() => StorageChild {
+						data: map![
+							vec![10] => vec![10],
+							vec![20] => vec![20],
+							vec![40] => vec![40]
+						],
+						child_info: child_info.to_owned(),
+					}
+				],
+			},
+			StateVersion::default(),
+		)
+			.into();

 		let ext = TestExt::new(&mut overlay, &mut cache, &backend, None);

@@ -993,20 +1024,23 @@ mod tests {
 		let mut overlay = OverlayedChanges::default();
 		overlay.set_child_storage(child_info, vec![20], None);
 		overlay.set_child_storage(child_info, vec![30], Some(vec![31]));
-		let backend = Storage {
-			top: map![],
-			children_default: map![
-				child_info.storage_key().to_vec() => StorageChild {
-					data: map![
-						vec![10] => vec![10],
-						vec![20] => vec![20],
-						vec![30] => vec![40]
-					],
-					child_info: child_info.to_owned(),
-				}
-			],
-		}
-		.into();
+		let backend = (
+			Storage {
+				top: map![],
+				children_default: map![
+					child_info.storage_key().to_vec() => StorageChild {
+						data: map![
+							vec![10] => vec![10],
+							vec![20] => vec![20],
+							vec![30] => vec![40]
+						],
+						child_info: child_info.to_owned(),
+					}
+				],
+			},
+			StateVersion::default(),
+		)
+			.into();

 		let ext = TestExt::new(&mut overlay, &mut cache, &backend, None);

@@ -1032,18 +1066,21 @@ mod tests {
 		let child_info = &child_info;
 		let mut cache = StorageTransactionCache::default();
 		let mut overlay = OverlayedChanges::default();
-		let backend = Storage {
-			top: map![],
-			children_default: map![
-				child_info.storage_key().to_vec() => StorageChild {
-					data: map![
-						vec![30] => vec![40]
-					],
-					child_info: child_info.to_owned(),
-				}
-			],
-		}
-		.into();
+		let backend = (
+			Storage {
+				top: map![],
+				children_default: map![
+					child_info.storage_key().to_vec() => StorageChild {
+						data: map![
+							vec![30] => vec![40]
+						],
+						child_info: child_info.to_owned(),
+					}
+				],
+			},
+			StateVersion::default(),
+		)
+			.into();

 		let ext = TestExt::new(&mut overlay, &mut cache, &backend, None);

@@ -22,8 +22,8 @@ use crate::{
 };
 use codec::Codec;
 use hash_db::Hasher;
-use sp_core::storage::{ChildInfo, Storage};
-use sp_trie::{empty_trie_root, Layout, MemoryDB};
+use sp_core::storage::{ChildInfo, StateVersion, Storage};
+use sp_trie::{empty_trie_root, LayoutV1, MemoryDB};
 use std::collections::{BTreeMap, HashMap};

 /// Create a new empty instance of in-memory backend.
@@ -32,7 +32,8 @@ where
 	H::Out: Codec + Ord,
 {
 	let db = MemoryDB::default();
-	TrieBackend::new(db, empty_trie_root::<Layout<H>>())
+	// V1 is same as V0 for an empty trie.
+	TrieBackend::new(db, empty_trie_root::<LayoutV1<H>>())
 }

 impl<H: Hasher> TrieBackend<MemoryDB<H>, H>
@@ -43,9 +44,10 @@ where
 	pub fn update<T: IntoIterator<Item = (Option<ChildInfo>, StorageCollection)>>(
 		&self,
 		changes: T,
+		state_version: StateVersion,
 	) -> Self {
 		let mut clone = self.clone();
-		clone.insert(changes);
+		clone.insert(changes, state_version);
 		clone
 	}

@@ -53,6 +55,7 @@ where
 	pub fn insert<T: IntoIterator<Item = (Option<ChildInfo>, StorageCollection)>>(
 		&mut self,
 		changes: T,
+		state_version: StateVersion,
 	) {
 		let (top, child) = changes.into_iter().partition::<Vec<_>, _>(|v| v.0.is_none());
 		let (root, transaction) = self.full_storage_root(
@@ -60,6 +63,7 @@ where
 			child.iter().filter_map(|v| {
 				v.0.as_ref().map(|c| (c, v.1.iter().map(|(k, v)| (&k[..], v.as_deref()))))
 			}),
+			state_version,
 		);

 		self.apply_transaction(root, transaction);
@@ -103,53 +107,63 @@ where
 	}
 }

-impl<H: Hasher> From<HashMap<Option<ChildInfo>, BTreeMap<StorageKey, StorageValue>>>
+impl<H: Hasher> From<(HashMap<Option<ChildInfo>, BTreeMap<StorageKey, StorageValue>>, StateVersion)>
 	for TrieBackend<MemoryDB<H>, H>
 where
 	H::Out: Codec + Ord,
 {
-	fn from(inner: HashMap<Option<ChildInfo>, BTreeMap<StorageKey, StorageValue>>) -> Self {
+	fn from(
+		(inner, state_version): (
+			HashMap<Option<ChildInfo>, BTreeMap<StorageKey, StorageValue>>,
+			StateVersion,
+		),
+	) -> Self {
 		let mut backend = new_in_mem();
 		backend.insert(
 			inner
 				.into_iter()
 				.map(|(k, m)| (k, m.into_iter().map(|(k, v)| (k, Some(v))).collect())),
+			state_version,
 		);
 		backend
 	}
 }

-impl<H: Hasher> From<Storage> for TrieBackend<MemoryDB<H>, H>
+impl<H: Hasher> From<(Storage, StateVersion)> for TrieBackend<MemoryDB<H>, H>
 where
 	H::Out: Codec + Ord,
 {
-	fn from(inners: Storage) -> Self {
+	fn from((inners, state_version): (Storage, StateVersion)) -> Self {
 		let mut inner: HashMap<Option<ChildInfo>, BTreeMap<StorageKey, StorageValue>> = inners
 			.children_default
 			.into_iter()
 			.map(|(_k, c)| (Some(c.child_info), c.data))
 			.collect();
 		inner.insert(None, inners.top);
-		inner.into()
+		(inner, state_version).into()
 	}
 }

-impl<H: Hasher> From<BTreeMap<StorageKey, StorageValue>> for TrieBackend<MemoryDB<H>, H>
+impl<H: Hasher> From<(BTreeMap<StorageKey, StorageValue>, StateVersion)>
+	for TrieBackend<MemoryDB<H>, H>
 where
 	H::Out: Codec + Ord,
 {
-	fn from(inner: BTreeMap<StorageKey, StorageValue>) -> Self {
+	fn from((inner, state_version): (BTreeMap<StorageKey, StorageValue>, StateVersion)) -> Self {
 		let mut expanded = HashMap::new();
 		expanded.insert(None, inner);
-		expanded.into()
+		(expanded, state_version).into()
 	}
 }

-impl<H: Hasher> From<Vec<(Option<ChildInfo>, StorageCollection)>> for TrieBackend<MemoryDB<H>, H>
+impl<H: Hasher> From<(Vec<(Option<ChildInfo>, StorageCollection)>, StateVersion)>
+	for TrieBackend<MemoryDB<H>, H>
 where
 	H::Out: Codec + Ord,
 {
-	fn from(inner: Vec<(Option<ChildInfo>, StorageCollection)>) -> Self {
+	fn from(
+		(inner, state_version): (Vec<(Option<ChildInfo>, StorageCollection)>, StateVersion),
+	) -> Self {
 		let mut expanded: HashMap<Option<ChildInfo>, BTreeMap<StorageKey, StorageValue>> =
 			HashMap::new();
 		for (child_info, key_values) in inner {
@@ -160,7 +174,7 @@ where
 				}
 			}
 		}
-		expanded.into()
+		(expanded, state_version).into()
 	}
 }

@@ -168,16 +182,20 @@ where
 mod tests {
 	use super::*;
 	use crate::backend::Backend;
+	use sp_core::storage::StateVersion;
 	use sp_runtime::traits::BlakeTwo256;

 	/// Assert in memory backend with only child trie keys works as trie backend.
 	#[test]
 	fn in_memory_with_child_trie_only() {
+		let state_version = StateVersion::default();
 		let storage = new_in_mem::<BlakeTwo256>();
 		let child_info = ChildInfo::new_default(b"1");
 		let child_info = &child_info;
-		let storage = storage
-			.update(vec![(Some(child_info.clone()), vec![(b"2".to_vec(), Some(b"3".to_vec()))])]);
+		let storage = storage.update(
+			vec![(Some(child_info.clone()), vec![(b"2".to_vec(), Some(b"3".to_vec()))])],
+			state_version,
+		);
 		let trie_backend = storage.as_trie_backend().unwrap();
 		assert_eq!(trie_backend.child_storage(child_info, b"2").unwrap(), Some(b"3".to_vec()));
 		let storage_key = child_info.prefixed_storage_key();
@@ -186,13 +204,18 @@ mod tests {

 	#[test]
 	fn insert_multiple_times_child_data_works() {
+		let state_version = StateVersion::default();
 		let mut storage = new_in_mem::<BlakeTwo256>();
 		let child_info = ChildInfo::new_default(b"1");

-		storage
-			.insert(vec![(Some(child_info.clone()), vec![(b"2".to_vec(), Some(b"3".to_vec()))])]);
-		storage
-			.insert(vec![(Some(child_info.clone()), vec![(b"1".to_vec(), Some(b"3".to_vec()))])]);
+		storage.insert(
+			vec![(Some(child_info.clone()), vec![(b"2".to_vec(), Some(b"3".to_vec()))])],
+			state_version,
+		);
+		storage.insert(
+			vec![(Some(child_info.clone()), vec![(b"1".to_vec(), Some(b"3".to_vec()))])],
+			state_version,
+		);

 		assert_eq!(storage.child_storage(&child_info, &b"2"[..]), Ok(Some(b"3".to_vec())));
 		assert_eq!(storage.child_storage(&child_info, &b"1"[..]), Ok(Some(b"3".to_vec())));
@@ -126,6 +126,7 @@ impl sp_std::fmt::Display for DefaultError {

 pub use crate::{
 	backend::Backend,
+	error::{Error, ExecutionError},
 	ext::Ext,
 	overlayed_changes::{
 		ChildStorageCollection, IndexOperation, OffchainChangesCollection,
@@ -136,7 +137,6 @@ pub use crate::{
 	trie_backend::TrieBackend,
 	trie_backend_essence::{Storage, TrieBackendStorage},
 };
-pub use error::{Error, ExecutionError};

 #[cfg(feature = "std")]
 mod std_reexport {
@@ -151,8 +151,8 @@ mod std_reexport {
 		testing::TestExternalities,
 	};
 	pub use sp_trie::{
-		trie_types::{Layout, TrieDBMut},
-		CompactProof, DBValue, MemoryDB, StorageProof, TrieMut,
+		trie_types::{TrieDBMutV0, TrieDBMutV1},
+		CompactProof, DBValue, LayoutV0, LayoutV1, MemoryDB, StorageProof, TrieMut,
 	};
 }

@@ -1353,7 +1353,7 @@ mod tests {
 	use codec::{Decode, Encode};
 	use sp_core::{
 		map,
-		storage::ChildInfo,
+		storage::{ChildInfo, StateVersion},
 		testing::TaskExecutor,
 		traits::{CodeExecutor, Externalities, RuntimeCode},
 		NativeOrEncoded, NeverNativeValue,
@@ -1416,7 +1416,11 @@ mod tests {

 	#[test]
 	fn execute_works() {
-		let backend = trie_backend::tests::test_trie();
+		execute_works_inner(StateVersion::V0);
+		execute_works_inner(StateVersion::V1);
+	}
+	fn execute_works_inner(state_version: StateVersion) {
+		let backend = trie_backend::tests::test_trie(state_version);
 		let mut overlayed_changes = Default::default();
 		let wasm_code = RuntimeCode::empty();

@@ -1440,7 +1444,11 @@ mod tests {

 	#[test]
 	fn execute_works_with_native_else_wasm() {
-		let backend = trie_backend::tests::test_trie();
+		execute_works_with_native_else_wasm_inner(StateVersion::V0);
+		execute_works_with_native_else_wasm_inner(StateVersion::V1);
+	}
+	fn execute_works_with_native_else_wasm_inner(state_version: StateVersion) {
+		let backend = trie_backend::tests::test_trie(state_version);
 		let mut overlayed_changes = Default::default();
 		let wasm_code = RuntimeCode::empty();

@@ -1464,8 +1472,12 @@ mod tests {

 	#[test]
 	fn dual_execution_strategy_detects_consensus_failure() {
+		dual_execution_strategy_detects_consensus_failure_inner(StateVersion::V0);
+		dual_execution_strategy_detects_consensus_failure_inner(StateVersion::V1);
+	}
+	fn dual_execution_strategy_detects_consensus_failure_inner(state_version: StateVersion) {
 		let mut consensus_failed = false;
-		let backend = trie_backend::tests::test_trie();
+		let backend = trie_backend::tests::test_trie(state_version);
 		let mut overlayed_changes = Default::default();
 		let wasm_code = RuntimeCode::empty();

@@ -1498,6 +1510,10 @@ mod tests {

 	#[test]
 	fn prove_execution_and_proof_check_works() {
+		prove_execution_and_proof_check_works_inner(StateVersion::V0);
+		prove_execution_and_proof_check_works_inner(StateVersion::V1);
+	}
+	fn prove_execution_and_proof_check_works_inner(state_version: StateVersion) {
 		let executor = DummyCodeExecutor {
 			native_available: true,
 			native_succeeds: true,
@@ -1505,8 +1521,8 @@ mod tests {
 		};

 		// fetch execution proof from 'remote' full node
-		let mut remote_backend = trie_backend::tests::test_trie();
-		let remote_root = remote_backend.storage_root(std::iter::empty()).0;
+		let mut remote_backend = trie_backend::tests::test_trie(state_version);
+		let remote_root = remote_backend.storage_root(std::iter::empty(), state_version).0;
 		let (remote_result, remote_proof) = prove_execution(
 			&mut remote_backend,
 			&mut Default::default(),
@@ -1544,7 +1560,7 @@ mod tests {
 			b"abc".to_vec() => b"2".to_vec(),
 			b"bbb".to_vec() => b"3".to_vec()
 		];
-		let state = InMemoryBackend::<BlakeTwo256>::from(initial);
+		let state = InMemoryBackend::<BlakeTwo256>::from((initial, StateVersion::default()));
 		let backend = state.as_trie_backend().unwrap();

 		let mut overlay = OverlayedChanges::default();
@@ -1613,7 +1629,7 @@ mod tests {
 				b"d".to_vec() => b"3".to_vec()
 			],
 		];
-		let backend = InMemoryBackend::<BlakeTwo256>::from(initial);
+		let backend = InMemoryBackend::<BlakeTwo256>::from((initial, StateVersion::default()));

 		let mut overlay = OverlayedChanges::default();
 		overlay.set_child_storage(&child_info, b"1".to_vec(), Some(b"1312".to_vec()));
@@ -1655,7 +1671,7 @@ mod tests {
 				b"d".to_vec() => b"3".to_vec()
 			],
 		];
-		let backend = InMemoryBackend::<BlakeTwo256>::from(initial);
+		let backend = InMemoryBackend::<BlakeTwo256>::from((initial, StateVersion::default()));
 		let mut overlay = OverlayedChanges::default();
 		let mut cache = StorageTransactionCache::default();
 		let mut ext = Ext::new(&mut overlay, &mut cache, &backend, None);
@@ -1789,13 +1805,17 @@ mod tests {

 	#[test]
 	fn prove_read_and_proof_check_works() {
+		prove_read_and_proof_check_works_inner(StateVersion::V0);
+		prove_read_and_proof_check_works_inner(StateVersion::V1);
+	}
+	fn prove_read_and_proof_check_works_inner(state_version: StateVersion) {
 		let child_info = ChildInfo::new_default(b"sub1");
 		let missing_child_info = ChildInfo::new_default(b"sub1sub2"); // key will include other child root to proof.
 		let child_info = &child_info;
 		let missing_child_info = &missing_child_info;
 		// fetch read proof from 'remote' full node
-		let remote_backend = trie_backend::tests::test_trie();
-		let remote_root = remote_backend.storage_root(std::iter::empty()).0;
+		let remote_backend = trie_backend::tests::test_trie(state_version);
+		let remote_root = remote_backend.storage_root(std::iter::empty(), state_version).0;
 		let remote_proof = prove_read(remote_backend, &[b"value2"]).unwrap();
 		let remote_proof = test_compact(remote_proof, &remote_root);
 		// check proof locally
@@ -1812,8 +1832,8 @@ mod tests {
 		);
 		assert_eq!(local_result2, false);
 		// on child trie
-		let remote_backend = trie_backend::tests::test_trie();
-		let remote_root = remote_backend.storage_root(std::iter::empty()).0;
+		let remote_backend = trie_backend::tests::test_trie(state_version);
+		let remote_root = remote_backend.storage_root(std::iter::empty(), state_version).0;
 		let remote_proof = prove_child_read(remote_backend, child_info, &[b"value3"]).unwrap();
 		let remote_proof = test_compact(remote_proof, &remote_root);
 		let local_result1 = read_child_proof_check::<BlakeTwo256, _>(
@@ -1877,7 +1897,8 @@ mod tests {
 				storage.insert(Some(child_info), items);
 			}

-			let trie: InMemoryBackend<BlakeTwo256> = storage.clone().into();
+			let trie: InMemoryBackend<BlakeTwo256> =
+				(storage.clone(), StateVersion::default()).into();
 			let trie_root = trie.root().clone();
 			let backend = crate::ProvingBackend::new(&trie);
 			let mut queries = Vec::new();
@@ -1940,15 +1961,16 @@ mod tests {

 	#[test]
 	fn prove_read_with_size_limit_works() {
-		let remote_backend = trie_backend::tests::test_trie();
-		let remote_root = remote_backend.storage_root(::std::iter::empty()).0;
+		let state_version = StateVersion::V0;
+		let remote_backend = trie_backend::tests::test_trie(state_version);
+		let remote_root = remote_backend.storage_root(::std::iter::empty(), state_version).0;
 		let (proof, count) =
 			prove_range_read_with_size(remote_backend, None, None, 0, None).unwrap();
 		// Always contains at least some nodes.
 		assert_eq!(proof.into_memory_db::<BlakeTwo256>().drain().len(), 3);
 		assert_eq!(count, 1);

-		let remote_backend = trie_backend::tests::test_trie();
+		let remote_backend = trie_backend::tests::test_trie(state_version);
 		let (proof, count) =
 			prove_range_read_with_size(remote_backend, None, None, 800, Some(&[])).unwrap();
 		assert_eq!(proof.clone().into_memory_db::<BlakeTwo256>().drain().len(), 9);
@@ -1971,7 +1993,7 @@ mod tests {
 		assert_eq!(results.len() as u32, 101);
 		assert_eq!(completed, false);

-		let remote_backend = trie_backend::tests::test_trie();
+		let remote_backend = trie_backend::tests::test_trie(state_version);
 		let (proof, count) =
 			prove_range_read_with_size(remote_backend, None, None, 50000, Some(&[])).unwrap();
 		assert_eq!(proof.clone().into_memory_db::<BlakeTwo256>().drain().len(), 11);
@@ -1989,10 +2011,66 @@ mod tests {
 		assert_eq!(completed, true);
 	}

+	#[test]
+	fn inner_state_versioning_switch_proofs() {
+		let mut state_version = StateVersion::V0;
+		let (mut mdb, mut root) = trie_backend::tests::test_db(state_version);
+		{
+			let mut trie = TrieDBMutV0::from_existing(&mut mdb, &mut root).unwrap();
+			trie.insert(b"foo", vec![1u8; 1_000].as_slice()) // big inner hash
+				.expect("insert failed");
+			trie.insert(b"foo2", vec![3u8; 16].as_slice()) // no inner hash
+				.expect("insert failed");
+			trie.insert(b"foo222", vec![5u8; 100].as_slice()) // inner hash
+				.expect("insert failed");
+		}
+
+		let check_proof = |mdb, root, state_version| -> StorageProof {
+			let remote_backend = TrieBackend::new(mdb, root);
+			let remote_root = remote_backend.storage_root(std::iter::empty(), state_version).0;
+			let remote_proof = prove_read(remote_backend, &[b"foo222"]).unwrap();
+			// check proof locally
+			let local_result1 =
+				read_proof_check::<BlakeTwo256, _>(remote_root, remote_proof.clone(), &[b"foo222"])
+					.unwrap();
+			// check that results are correct
+			assert_eq!(
+				local_result1.into_iter().collect::<Vec<_>>(),
+				vec![(b"foo222".to_vec(), Some(vec![5u8; 100]))],
+			);
+			remote_proof
+		};
+
+		let remote_proof = check_proof(mdb.clone(), root.clone(), state_version);
+		// check full values in proof
+		assert!(remote_proof.encode().len() > 1_100);
+		assert!(remote_proof.encoded_size() > 1_100);
+		let root1 = root.clone();
+
+		// do switch
+		state_version = StateVersion::V1;
+		{
+			let mut trie = TrieDBMutV1::from_existing(&mut mdb, &mut root).unwrap();
+			trie.insert(b"foo222", vec![5u8; 100].as_slice()) // inner hash
+				.expect("insert failed");
+			// update with same value do change
+			trie.insert(b"foo", vec![1u8; 1000].as_slice()) // inner hash
+				.expect("insert failed");
+		}
+		let root3 = root.clone();
+		assert!(root1 != root3);
+		let remote_proof = check_proof(mdb.clone(), root.clone(), state_version);
+		// nodes foo is replaced by its hashed value form.
+		assert!(remote_proof.encode().len() < 1000);
+		assert!(remote_proof.encoded_size() < 1000);
+		assert_eq!(remote_proof.encode().len(), remote_proof.encoded_size());
+	}
+
 	#[test]
 	fn prove_range_with_child_works() {
-		let remote_backend = trie_backend::tests::test_trie();
-		let remote_root = remote_backend.storage_root(::std::iter::empty()).0;
+		let state_version = StateVersion::V0;
+		let remote_backend = trie_backend::tests::test_trie(state_version);
+		let remote_root = remote_backend.storage_root(std::iter::empty(), state_version).0;
 		let mut start_at = smallvec::SmallVec::<[Vec<u8>; 2]>::new();
 		let trie_backend = remote_backend.as_trie_backend().unwrap();
 		let max_iter = 1000;
@@ -2030,20 +2108,34 @@ mod tests {

 	#[test]
 	fn compact_multiple_child_trie() {
+		let size_no_inner_hash = compact_multiple_child_trie_inner(StateVersion::V0);
+		let size_inner_hash = compact_multiple_child_trie_inner(StateVersion::V1);
+		assert!(size_inner_hash < size_no_inner_hash);
+	}
+	fn compact_multiple_child_trie_inner(state_version: StateVersion) -> usize {
 		// this root will be queried
 		let child_info1 = ChildInfo::new_default(b"sub1");
 		// this root will not be include in proof
 		let child_info2 = ChildInfo::new_default(b"sub2");
 		// this root will be include in proof
 		let child_info3 = ChildInfo::new_default(b"sub");
-		let remote_backend = trie_backend::tests::test_trie();
+		let remote_backend = trie_backend::tests::test_trie(state_version);
+		let long_vec: Vec<u8> = (0..1024usize).map(|_| 8u8).collect();
 		let (remote_root, transaction) = remote_backend.full_storage_root(
 			std::iter::empty(),
 			vec![
 				(
 					&child_info1,
-					vec![(&b"key1"[..], Some(&b"val2"[..])), (&b"key2"[..], Some(&b"val3"[..]))]
-						.into_iter(),
+					vec![
+						// a inner hashable node
+						(&b"k"[..], Some(&long_vec[..])),
+						// need to ensure this is not an inline node
+						// otherwhise we do not know what is accessed when
+						// storing proof.
+						(&b"key1"[..], Some(&vec![5u8; 32][..])),
+						(&b"key2"[..], Some(&b"val3"[..])),
+					]
+					.into_iter(),
 				),
 				(
 					&child_info2,
@@ -2057,11 +2149,13 @@ mod tests {
 				),
 			]
 			.into_iter(),
+			state_version,
 		);
 		let mut remote_storage = remote_backend.into_storage();
 		remote_storage.consolidate(transaction);
 		let remote_backend = TrieBackend::new(remote_storage, remote_root);
 		let remote_proof = prove_child_read(remote_backend, &child_info1, &[b"key1"]).unwrap();
+		let size = remote_proof.encoded_size();
 		let remote_proof = test_compact(remote_proof, &remote_root);
 		let local_result1 = read_child_proof_check::<BlakeTwo256, _>(
 			remote_root,
@@ -2071,11 +2165,13 @@ mod tests {
 		)
 		.unwrap();
 		assert_eq!(local_result1.len(), 1);
-		assert_eq!(local_result1.get(&b"key1"[..]), Some(&Some(b"val2".to_vec())));
+		assert_eq!(local_result1.get(&b"key1"[..]), Some(&Some(vec![5u8; 32])));
+		size
 	}

 	#[test]
 	fn child_storage_uuid() {
+		let state_version = StateVersion::V0;
 		let child_info_1 = ChildInfo::new_default(b"sub_test1");
 		let child_info_2 = ChildInfo::new_default(b"sub_test2");

@@ -2083,12 +2179,12 @@ mod tests {
 		let mut overlay = OverlayedChanges::default();

 		let mut transaction = {
-			let backend = test_trie();
+			let backend = test_trie(state_version);
 			let mut cache = StorageTransactionCache::default();
 			let mut ext = Ext::new(&mut overlay, &mut cache, &backend, None);
 			ext.set_child_storage(&child_info_1, b"abc".to_vec(), b"def".to_vec());
 			ext.set_child_storage(&child_info_2, b"abc".to_vec(), b"def".to_vec());
-			ext.storage_root();
+			ext.storage_root(state_version);
 			cache.transaction.unwrap()
 		};
 		let mut duplicate = false;
@@ -2108,7 +2204,7 @@ mod tests {
 			b"aaa".to_vec() => b"0".to_vec(),
 			b"bbb".to_vec() => b"".to_vec()
 		];
-		let state = InMemoryBackend::<BlakeTwo256>::from(initial);
+		let state = InMemoryBackend::<BlakeTwo256>::from((initial, StateVersion::default()));
 		let backend = state.as_trie_backend().unwrap();

 		let mut overlay = OverlayedChanges::default();
@@ -2134,12 +2230,13 @@ mod tests {

 	#[test]
 	fn runtime_registered_extensions_are_removed_after_execution() {
+		let state_version = StateVersion::default();
 		use sp_externalities::ExternalitiesExt;
 		sp_externalities::decl_extension! {
 			struct DummyExt(u32);
 		}

-		let backend = trie_backend::tests::test_trie();
+		let backend = trie_backend::tests::test_trie(state_version);
 		let mut overlayed_changes = Default::default();
 		let wasm_code = RuntimeCode::empty();

@@ -27,7 +27,7 @@ use hash_db::Hasher;
 pub use offchain::OffchainOverlayedChanges;
 use sp_core::{
 	offchain::OffchainOverlayedChange,
-	storage::{well_known_keys::EXTRINSIC_INDEX, ChildInfo},
+	storage::{well_known_keys::EXTRINSIC_INDEX, ChildInfo, StateVersion},
 };
 #[cfg(feature = "std")]
 use sp_externalities::{Extension, Extensions};
@@ -502,11 +502,12 @@ impl OverlayedChanges {
 		backend: &B,
 		parent_hash: H::Out,
 		mut cache: StorageTransactionCache<B::Transaction, H>,
+		state_version: StateVersion,
 	) -> Result<StorageChanges<B::Transaction, H>, DefaultError>
 	where
 		H::Out: Ord + Encode + 'static,
 	{
-		self.drain_storage_changes(backend, parent_hash, &mut cache)
+		self.drain_storage_changes(backend, parent_hash, &mut cache, state_version)
 	}

 	/// Drain all changes into a [`StorageChanges`] instance. Leave empty overlay in place.
@@ -515,13 +516,14 @@ impl OverlayedChanges {
 		backend: &B,
 		_parent_hash: H::Out,
 		mut cache: &mut StorageTransactionCache<B::Transaction, H>,
+		state_version: StateVersion,
 	) -> Result<StorageChanges<B::Transaction, H>, DefaultError>
 	where
 		H::Out: Ord + Encode + 'static,
 	{
 		// If the transaction does not exist, we generate it.
 		if cache.transaction.is_none() {
-			self.storage_root(backend, &mut cache);
+			self.storage_root(backend, &mut cache, state_version);
 		}

 		let (transaction, transaction_storage_root) = cache
@@ -580,6 +582,7 @@ impl OverlayedChanges {
 		&self,
 		backend: &B,
 		cache: &mut StorageTransactionCache<B::Transaction, H>,
+		state_version: StateVersion,
 	) -> H::Out
 	where
 		H::Out: Ord + Encode,
@@ -589,7 +592,7 @@ impl OverlayedChanges {
 			(info, changes.map(|(k, v)| (&k[..], v.value().map(|v| &v[..]))))
 		});

-		let (root, transaction) = backend.full_storage_root(delta, child_delta);
+		let (root, transaction) = backend.full_storage_root(delta, child_delta, state_version);

 		cache.transaction = Some(transaction);
 		cache.transaction_storage_root = Some(root);
@@ -830,6 +833,7 @@ mod tests {

 	#[test]
 	fn overlayed_storage_root_works() {
+		let state_version = StateVersion::default();
 		let initial: BTreeMap<_, _> = vec![
 			(b"doe".to_vec(), b"reindeer".to_vec()),
 			(b"dog".to_vec(), b"puppyXXX".to_vec()),
@@ -838,7 +842,7 @@ mod tests {
 		]
 		.into_iter()
 		.collect();
-		let backend = InMemoryBackend::<Blake2Hasher>::from(initial);
+		let backend = InMemoryBackend::<Blake2Hasher>::from((initial, state_version));
 		let mut overlay = OverlayedChanges::default();

 		overlay.start_transaction();
@@ -856,7 +860,7 @@ mod tests {
 		const ROOT: [u8; 32] =
 			hex!("39245109cef3758c2eed2ccba8d9b370a917850af3824bc8348d505df2c298fa");

-		assert_eq!(&ext.storage_root()[..], &ROOT);
+		assert_eq!(&ext.storage_root(state_version)[..], &ROOT);
 	}

 	#[test]
@@ -26,14 +26,11 @@ use codec::{Codec, Decode, Encode};
 use hash_db::{HashDB, Hasher, Prefix, EMPTY_PREFIX};
 use log::debug;
 use parking_lot::RwLock;
-use sp_core::storage::ChildInfo;
+use sp_core::storage::{ChildInfo, StateVersion};
+pub use sp_trie::trie_types::TrieError;
 use sp_trie::{
 	empty_child_trie_root, read_child_trie_value_with, read_trie_value_with, record_all_keys,
-	MemoryDB, StorageProof,
-};
-pub use sp_trie::{
-	trie_types::{Layout, TrieError},
-	Recorder,
+	LayoutV1, MemoryDB, Recorder, StorageProof,
 };
 use std::{
 	collections::{hash_map::Entry, HashMap},
@@ -59,7 +56,8 @@ where

 		let map_e = |e| format!("Trie lookup error: {}", e);

-		read_trie_value_with::<Layout<H>, _, Ephemeral<S, H>>(
+		// V1 is equivalent to V0 on read.
+		read_trie_value_with::<LayoutV1<H>, _, Ephemeral<S, H>>(
 			&eph,
 			self.backend.root(),
 			key,
@@ -78,14 +76,16 @@ where
 		let root = self
 			.storage(storage_key)?
 			.and_then(|r| Decode::decode(&mut &r[..]).ok())
-			.unwrap_or_else(|| empty_child_trie_root::<Layout<H>>());
+			// V1 is equivalent to V0 on empty trie
+			.unwrap_or_else(|| empty_child_trie_root::<LayoutV1<H>>());

 		let mut read_overlay = S::Overlay::default();
 		let eph = Ephemeral::new(self.backend.backend_storage(), &mut read_overlay);

 		let map_e = |e| format!("Trie lookup error: {}", e);

-		read_child_trie_value_with::<Layout<H>, _, _>(
+		// V1 is equivalent to V0 on read
+		read_child_trie_value_with::<LayoutV1<H>, _, _>(
 			child_info.keyspace(),
 			&eph,
 			&root.as_ref(),
@@ -102,7 +102,8 @@ where

 		let mut iter = move || -> Result<(), Box<TrieError<H::Out>>> {
 			let root = self.backend.root();
-			record_all_keys::<Layout<H>, _>(&eph, root, &mut *self.proof_recorder)
+			// V1 and V is equivalent to V0 on read and recorder is key read.
+			record_all_keys::<LayoutV1<H>, _>(&eph, root, &mut *self.proof_recorder)
 		};

 		if let Err(e) = iter() {
@@ -338,22 +339,24 @@ where
 	fn storage_root<'b>(
 		&self,
 		delta: impl Iterator<Item = (&'b [u8], Option<&'b [u8]>)>,
+		state_version: StateVersion,
 	) -> (H::Out, Self::Transaction)
 	where
 		H::Out: Ord,
 	{
-		self.0.storage_root(delta)
+		self.0.storage_root(delta, state_version)
 	}

 	fn child_storage_root<'b>(
 		&self,
 		child_info: &ChildInfo,
 		delta: impl Iterator<Item = (&'b [u8], Option<&'b [u8]>)>,
+		state_version: StateVersion,
 	) -> (H::Out, bool, Self::Transaction)
 	where
 		H::Out: Ord,
 	{
-		self.0.child_storage_root(child_info, delta)
+		self.0.child_storage_root(child_info, delta, state_version)
 	}

 	fn register_overlay_stats(&self, _stats: &crate::stats::StateMachineStats) {}
@@ -401,13 +404,21 @@ mod tests {

 	#[test]
 	fn proof_is_empty_until_value_is_read() {
-		let trie_backend = test_trie();
+		proof_is_empty_until_value_is_read_inner(StateVersion::V0);
+		proof_is_empty_until_value_is_read_inner(StateVersion::V1);
+	}
+	fn proof_is_empty_until_value_is_read_inner(test_hash: StateVersion) {
+		let trie_backend = test_trie(test_hash);
 		assert!(test_proving(&trie_backend).extract_proof().is_empty());
 	}

 	#[test]
 	fn proof_is_non_empty_after_value_is_read() {
-		let trie_backend = test_trie();
+		proof_is_non_empty_after_value_is_read_inner(StateVersion::V0);
+		proof_is_non_empty_after_value_is_read_inner(StateVersion::V1);
+	}
+	fn proof_is_non_empty_after_value_is_read_inner(test_hash: StateVersion) {
+		let trie_backend = test_trie(test_hash);
 		let backend = test_proving(&trie_backend);
 		assert_eq!(backend.storage(b"key").unwrap(), Some(b"value".to_vec()));
 		assert!(!backend.extract_proof().is_empty());
@@ -425,58 +436,82 @@ mod tests {

 	#[test]
 	fn passes_through_backend_calls() {
-		let trie_backend = test_trie();
+		passes_through_backend_calls_inner(StateVersion::V0);
+		passes_through_backend_calls_inner(StateVersion::V1);
+	}
+	fn passes_through_backend_calls_inner(state_version: StateVersion) {
+		let trie_backend = test_trie(state_version);
 		let proving_backend = test_proving(&trie_backend);
 		assert_eq!(trie_backend.storage(b"key").unwrap(), proving_backend.storage(b"key").unwrap());
 		assert_eq!(trie_backend.pairs(), proving_backend.pairs());

-		let (trie_root, mut trie_mdb) = trie_backend.storage_root(std::iter::empty());
-		let (proving_root, mut proving_mdb) = proving_backend.storage_root(std::iter::empty());
+		let (trie_root, mut trie_mdb) =
+			trie_backend.storage_root(std::iter::empty(), state_version);
+		let (proving_root, mut proving_mdb) =
+			proving_backend.storage_root(std::iter::empty(), state_version);
 		assert_eq!(trie_root, proving_root);
 		assert_eq!(trie_mdb.drain(), proving_mdb.drain());
 	}

 	#[test]
-	fn proof_recorded_and_checked() {
-		let contents = (0..64).map(|i| (vec![i], Some(vec![i]))).collect::<Vec<_>>();
+	fn proof_recorded_and_checked_top() {
+		proof_recorded_and_checked_inner(StateVersion::V0);
+		proof_recorded_and_checked_inner(StateVersion::V1);
+	}
+	fn proof_recorded_and_checked_inner(state_version: StateVersion) {
+		let size_content = 34; // above hashable value treshold.
+		let value_range = 0..64;
+		let contents = value_range
+			.clone()
+			.map(|i| (vec![i], Some(vec![i; size_content])))
+			.collect::<Vec<_>>();
 		let in_memory = InMemoryBackend::<BlakeTwo256>::default();
-		let in_memory = in_memory.update(vec![(None, contents)]);
-		let in_memory_root = in_memory.storage_root(::std::iter::empty()).0;
-		(0..64).for_each(|i| assert_eq!(in_memory.storage(&[i]).unwrap().unwrap(), vec![i]));
+		let in_memory = in_memory.update(vec![(None, contents)], state_version);
+		let in_memory_root = in_memory.storage_root(std::iter::empty(), state_version).0;
+		value_range.clone().for_each(|i| {
+			assert_eq!(in_memory.storage(&[i]).unwrap().unwrap(), vec![i; size_content])
+		});

 		let trie = in_memory.as_trie_backend().unwrap();
-		let trie_root = trie.storage_root(::std::iter::empty()).0;
+		let trie_root = trie.storage_root(std::iter::empty(), state_version).0;
 		assert_eq!(in_memory_root, trie_root);
-		(0..64).for_each(|i| assert_eq!(trie.storage(&[i]).unwrap().unwrap(), vec![i]));
+		value_range
+			.clone()
+			.for_each(|i| assert_eq!(trie.storage(&[i]).unwrap().unwrap(), vec![i; size_content]));

 		let proving = ProvingBackend::new(trie);
-		assert_eq!(proving.storage(&[42]).unwrap().unwrap(), vec![42]);
+		assert_eq!(proving.storage(&[42]).unwrap().unwrap(), vec![42; size_content]);

 		let proof = proving.extract_proof();

 		let proof_check =
 			create_proof_check_backend::<BlakeTwo256>(in_memory_root.into(), proof).unwrap();
-		assert_eq!(proof_check.storage(&[42]).unwrap().unwrap(), vec![42]);
+		assert_eq!(proof_check.storage(&[42]).unwrap().unwrap(), vec![42; size_content]);
 	}

 	#[test]
 	fn proof_recorded_and_checked_with_child() {
+		proof_recorded_and_checked_with_child_inner(StateVersion::V0);
+		proof_recorded_and_checked_with_child_inner(StateVersion::V1);
+	}
+	fn proof_recorded_and_checked_with_child_inner(state_version: StateVersion) {
 		let child_info_1 = ChildInfo::new_default(b"sub1");
 		let child_info_2 = ChildInfo::new_default(b"sub2");
 		let child_info_1 = &child_info_1;
 		let child_info_2 = &child_info_2;
 		let contents = vec![
-			(None, (0..64).map(|i| (vec![i], Some(vec![i]))).collect()),
+			(None, (0..64).map(|i| (vec![i], Some(vec![i]))).collect::<Vec<_>>()),
 			(Some(child_info_1.clone()), (28..65).map(|i| (vec![i], Some(vec![i]))).collect()),
 			(Some(child_info_2.clone()), (10..15).map(|i| (vec![i], Some(vec![i]))).collect()),
 		];
 		let in_memory = InMemoryBackend::<BlakeTwo256>::default();
-		let in_memory = in_memory.update(contents);
+		let in_memory = in_memory.update(contents, state_version);
 		let child_storage_keys = vec![child_info_1.to_owned(), child_info_2.to_owned()];
 		let in_memory_root = in_memory
 			.full_storage_root(
 				std::iter::empty(),
 				child_storage_keys.iter().map(|k| (k, std::iter::empty())),
+				state_version,
 			)
 			.0;
 		(0..64).for_each(|i| assert_eq!(in_memory.storage(&[i]).unwrap().unwrap(), vec![i]));
@@ -488,7 +523,7 @@ mod tests {
 		});

 		let trie = in_memory.as_trie_backend().unwrap();
-		let trie_root = trie.storage_root(std::iter::empty()).0;
+		let trie_root = trie.storage_root(std::iter::empty(), state_version).0;
 		assert_eq!(in_memory_root, trie_root);
 		(0..64).for_each(|i| assert_eq!(trie.storage(&[i]).unwrap().unwrap(), vec![i]));

@@ -516,7 +551,11 @@ mod tests {

 	#[test]
 	fn storage_proof_encoded_size_estimation_works() {
-		let trie_backend = test_trie();
+		storage_proof_encoded_size_estimation_works_inner(StateVersion::V0);
+		storage_proof_encoded_size_estimation_works_inner(StateVersion::V1);
+	}
+	fn storage_proof_encoded_size_estimation_works_inner(state_version: StateVersion) {
+		let trie_backend = test_trie(state_version);
 		let backend = test_proving(&trie_backend);

 		let check_estimation =
@@ -21,7 +21,7 @@ use crate::{Backend, StorageKey, StorageValue};
 use codec::Encode;
 use hash_db::Hasher;
 use sp_core::{
-	storage::{ChildInfo, TrackedStorageKey},
+	storage::{ChildInfo, StateVersion, TrackedStorageKey},
 	traits::Externalities,
 	Blake2Hasher,
 };
@@ -145,11 +145,15 @@ impl<'a, H: Hasher, B: 'a + Backend<H>> Externalities for ReadOnlyExternalities<
 		unimplemented!("storage_append is not supported in ReadOnlyExternalities")
 	}

-	fn storage_root(&mut self) -> Vec<u8> {
+	fn storage_root(&mut self, _state_version: StateVersion) -> Vec<u8> {
 		unimplemented!("storage_root is not supported in ReadOnlyExternalities")
 	}

-	fn child_storage_root(&mut self, _child_info: &ChildInfo) -> Vec<u8> {
+	fn child_storage_root(
+		&mut self,
+		_child_info: &ChildInfo,
+		_state_version: StateVersion,
+	) -> Vec<u8> {
 		unimplemented!("child_storage_root is not supported in ReadOnlyExternalities")
 	}

@@ -32,7 +32,7 @@ use sp_core::{
 	offchain::testing::TestPersistentOffchainDB,
 	storage::{
 		well_known_keys::{is_child_storage_key, CODE},
-		Storage,
+		StateVersion, Storage,
 	},
 	testing::TaskExecutor,
 	traits::TaskExecutorExt,
@@ -54,6 +54,8 @@ where
 	pub backend: InMemoryBackend<H>,
 	/// Extensions.
 	pub extensions: Extensions,
+	/// State version to use during tests.
+	pub state_version: StateVersion,
 }

 impl<H: Hasher> TestExternalities<H>
@@ -72,18 +74,31 @@ where

 	/// Create a new instance of `TestExternalities` with storage.
 	pub fn new(storage: Storage) -> Self {
-		Self::new_with_code(&[], storage)
+		Self::new_with_code_and_state(&[], storage, Default::default())
+	}
+
+	/// Create a new instance of `TestExternalities` with storage for a given state version.
+	pub fn new_with_state_version(storage: Storage, state_version: StateVersion) -> Self {
+		Self::new_with_code_and_state(&[], storage, state_version)
 	}

 	/// New empty test externalities.
 	pub fn new_empty() -> Self {
-		Self::new_with_code(&[], Storage::default())
+		Self::new_with_code_and_state(&[], Storage::default(), Default::default())
 	}

 	/// Create a new instance of `TestExternalities` with code and storage.
-	pub fn new_with_code(code: &[u8], mut storage: Storage) -> Self {
-		let overlay = OverlayedChanges::default();
+	pub fn new_with_code(code: &[u8], storage: Storage) -> Self {
+		Self::new_with_code_and_state(code, storage, Default::default())
+	}

+	/// Create a new instance of `TestExternalities` with code and storage for a given state
+	/// version.
+	pub fn new_with_code_and_state(
+		code: &[u8],
+		mut storage: Storage,
+		state_version: StateVersion,
+	) -> Self {
 		assert!(storage.top.keys().all(|key| !is_child_storage_key(key)));
 		assert!(storage.children_default.keys().all(|key| is_child_storage_key(key)));

@@ -94,12 +109,15 @@ where

 		let offchain_db = TestPersistentOffchainDB::new();

+		let backend = (storage, state_version).into();
+
 		TestExternalities {
-			overlay,
+			overlay: OverlayedChanges::default(),
 			offchain_db,
 			extensions,
-			backend: storage.into(),
+			backend,
 			storage_transaction_cache: Default::default(),
+			state_version,
 		}
 	}

@@ -120,14 +138,14 @@ where

 	/// Insert key/value into backend
 	pub fn insert(&mut self, k: StorageKey, v: StorageValue) {
-		self.backend.insert(vec![(None, vec![(k, Some(v))])]);
+		self.backend.insert(vec![(None, vec![(k, Some(v))])], self.state_version);
 	}

 	/// Insert key/value into backend.
 	///
 	/// This only supports inserting keys in child tries.
 	pub fn insert_child(&mut self, c: sp_core::storage::ChildInfo, k: StorageKey, v: StorageValue) {
-		self.backend.insert(vec![(Some(c), vec![(k, Some(v))])]);
+		self.backend.insert(vec![(Some(c), vec![(k, Some(v))])], self.state_version);
 	}

 	/// Registers the given extension for this instance.
@@ -151,7 +169,7 @@ where
 			))
 		}

-		self.backend.update(transaction)
+		self.backend.update(transaction, self.state_version)
 	}

 	/// Commit all pending changes to the underlying backend.
@@ -164,6 +182,7 @@ where
 			&self.backend,
 			Default::default(),
 			&mut Default::default(),
+			self.state_version,
 		)?;

 		self.backend
@@ -240,7 +259,8 @@ where
 	H::Out: Ord + 'static + codec::Codec,
 {
 	fn default() -> Self {
-		Self::new(Default::default())
+		// default to default version.
+		Self::new_with_state_version(Storage::default(), Default::default())
 	}
 }

@@ -249,7 +269,16 @@ where
 	H::Out: Ord + 'static + codec::Codec,
 {
 	fn from(storage: Storage) -> Self {
-		Self::new(storage)
+		Self::new_with_state_version(storage, Default::default())
+	}
+}
+
+impl<H: Hasher> From<(Storage, StateVersion)> for TestExternalities<H>
+where
+	H::Out: Ord + 'static + codec::Codec,
+{
+	fn from((storage, state_version): (Storage, StateVersion)) -> Self {
+		Self::new_with_state_version(storage, state_version)
 	}
 }

@@ -309,14 +338,15 @@ mod tests {

 	#[test]
 	fn commit_should_work() {
-		let mut ext = TestExternalities::<BlakeTwo256>::default();
+		let storage = Storage::default(); // avoid adding the trie threshold.
+		let mut ext = TestExternalities::<BlakeTwo256>::from((storage, Default::default()));
 		let mut ext = ext.ext();
 		ext.set_storage(b"doe".to_vec(), b"reindeer".to_vec());
 		ext.set_storage(b"dog".to_vec(), b"puppy".to_vec());
 		ext.set_storage(b"dogglesworth".to_vec(), b"cat".to_vec());
 		let root =
 			H256::from(hex!("ed4d8c799d996add422395a6abd7545491d40bd838d738afafa1b8a4de625489"));
-		assert_eq!(H256::from_slice(ext.storage_root().as_slice()), root);
+		assert_eq!(H256::from_slice(ext.storage_root(Default::default()).as_slice()), root);
 	}

 	#[test]
@@ -24,12 +24,12 @@ use crate::{
 };
 use codec::{Codec, Decode};
 use hash_db::Hasher;
-use sp_core::storage::{ChildInfo, ChildType};
+use sp_core::storage::{ChildInfo, ChildType, StateVersion};
 use sp_std::{boxed::Box, vec::Vec};
 use sp_trie::{
 	child_delta_trie_root, delta_trie_root, empty_child_trie_root,
-	trie_types::{Layout, TrieDB, TrieError},
-	Trie,
+	trie_types::{TrieDB, TrieError},
+	LayoutV0, LayoutV1, Trie,
 };

 /// Patricia trie-based backend. Transaction type is an overlay of changes to commit.
@@ -186,6 +186,7 @@ where
 	fn storage_root<'a>(
 		&self,
 		delta: impl Iterator<Item = (&'a [u8], Option<&'a [u8]>)>,
+		state_version: StateVersion,
 	) -> (H::Out, Self::Transaction)
 	where
 		H::Out: Ord,
@@ -195,8 +196,14 @@ where

 		{
 			let mut eph = Ephemeral::new(self.essence.backend_storage(), &mut write_overlay);
+			let res = match state_version {
+				StateVersion::V0 =>
+					delta_trie_root::<LayoutV0<H>, _, _, _, _, _>(&mut eph, root, delta),
+				StateVersion::V1 =>
+					delta_trie_root::<LayoutV1<H>, _, _, _, _, _>(&mut eph, root, delta),
+			};

-			match delta_trie_root::<Layout<H>, _, _, _, _, _>(&mut eph, root, delta) {
+			match res {
 				Ok(ret) => root = ret,
 				Err(e) => warn!(target: "trie", "Failed to write to trie: {}", e),
 			}
@@ -209,14 +216,14 @@ where
 		&self,
 		child_info: &ChildInfo,
 		delta: impl Iterator<Item = (&'a [u8], Option<&'a [u8]>)>,
+		state_version: StateVersion,
 	) -> (H::Out, bool, Self::Transaction)
 	where
 		H::Out: Ord,
 	{
 		let default_root = match child_info.child_type() {
-			ChildType::ParentKeyId => empty_child_trie_root::<Layout<H>>(),
+			ChildType::ParentKeyId => empty_child_trie_root::<LayoutV1<H>>(),
 		};
-
 		let mut write_overlay = S::Overlay::default();
 		let prefixed_storage_key = child_info.prefixed_storage_key();
 		let mut root = match self.storage(prefixed_storage_key.as_slice()) {
@@ -231,13 +238,20 @@ where

 		{
 			let mut eph = Ephemeral::new(self.essence.backend_storage(), &mut write_overlay);
-
-			match child_delta_trie_root::<Layout<H>, _, _, _, _, _, _>(
-				child_info.keyspace(),
-				&mut eph,
-				root,
-				delta,
-			) {
+			match match state_version {
+				StateVersion::V0 => child_delta_trie_root::<LayoutV0<H>, _, _, _, _, _, _>(
+					child_info.keyspace(),
+					&mut eph,
+					root,
+					delta,
+				),
+				StateVersion::V1 => child_delta_trie_root::<LayoutV1<H>, _, _, _, _, _, _>(
+					child_info.keyspace(),
+					&mut eph,
+					root,
+					delta,
+				),
+			} {
 				Ok(ret) => root = ret,
 				Err(e) => warn!(target: "trie", "Failed to write to trie: {}", e),
 			}
@@ -269,52 +283,91 @@ pub mod tests {
 	use codec::Encode;
 	use sp_core::H256;
 	use sp_runtime::traits::BlakeTwo256;
-	use sp_trie::{trie_types::TrieDBMut, KeySpacedDBMut, PrefixedMemoryDB, TrieMut};
+	use sp_trie::{
+		trie_types::{TrieDBMutV0, TrieDBMutV1},
+		KeySpacedDBMut, PrefixedMemoryDB, TrieMut,
+	};
 	use std::{collections::HashSet, iter};

 	const CHILD_KEY_1: &[u8] = b"sub1";

-	fn test_db() -> (PrefixedMemoryDB<BlakeTwo256>, H256) {
+	pub(crate) fn test_db(state_version: StateVersion) -> (PrefixedMemoryDB<BlakeTwo256>, H256) {
 		let child_info = ChildInfo::new_default(CHILD_KEY_1);
 		let mut root = H256::default();
 		let mut mdb = PrefixedMemoryDB::<BlakeTwo256>::default();
 		{
 			let mut mdb = KeySpacedDBMut::new(&mut mdb, child_info.keyspace());
-			let mut trie = TrieDBMut::new(&mut mdb, &mut root);
-			trie.insert(b"value3", &[142; 33]).expect("insert failed");
-			trie.insert(b"value4", &[124; 33]).expect("insert failed");
+			match state_version {
+				StateVersion::V0 => {
+					let mut trie = TrieDBMutV0::new(&mut mdb, &mut root);
+					trie.insert(b"value3", &[142; 33]).expect("insert failed");
+					trie.insert(b"value4", &[124; 33]).expect("insert failed");
+				},
+				StateVersion::V1 => {
+					let mut trie = TrieDBMutV1::new(&mut mdb, &mut root);
+					trie.insert(b"value3", &[142; 33]).expect("insert failed");
+					trie.insert(b"value4", &[124; 33]).expect("insert failed");
+				},
+			};
 		};

 		{
 			let mut sub_root = Vec::new();
 			root.encode_to(&mut sub_root);
-			let mut trie = TrieDBMut::new(&mut mdb, &mut root);
-			trie.insert(child_info.prefixed_storage_key().as_slice(), &sub_root[..])
-				.expect("insert failed");
-			trie.insert(b"key", b"value").expect("insert failed");
-			trie.insert(b"value1", &[42]).expect("insert failed");
-			trie.insert(b"value2", &[24]).expect("insert failed");
-			trie.insert(b":code", b"return 42").expect("insert failed");
-			for i in 128u8..255u8 {
-				trie.insert(&[i], &[i]).unwrap();
+
+			fn build<L: sp_trie::TrieLayout>(
+				mut trie: sp_trie::TrieDBMut<L>,
+				child_info: &ChildInfo,
+				sub_root: &[u8],
+			) {
+				trie.insert(child_info.prefixed_storage_key().as_slice(), sub_root)
+					.expect("insert failed");
+				trie.insert(b"key", b"value").expect("insert failed");
+				trie.insert(b"value1", &[42]).expect("insert failed");
+				trie.insert(b"value2", &[24]).expect("insert failed");
+				trie.insert(b":code", b"return 42").expect("insert failed");
+				for i in 128u8..255u8 {
+					trie.insert(&[i], &[i]).unwrap();
+				}
 			}
+
+			match state_version {
+				StateVersion::V0 => {
+					let trie = TrieDBMutV0::new(&mut mdb, &mut root);
+					build(trie, &child_info, &sub_root[..])
+				},
+				StateVersion::V1 => {
+					let trie = TrieDBMutV1::new(&mut mdb, &mut root);
+					build(trie, &child_info, &sub_root[..])
+				},
+			};
 		}
 		(mdb, root)
 	}

-	pub(crate) fn test_trie() -> TrieBackend<PrefixedMemoryDB<BlakeTwo256>, BlakeTwo256> {
-		let (mdb, root) = test_db();
+	pub(crate) fn test_trie(
+		hashed_value: StateVersion,
+	) -> TrieBackend<PrefixedMemoryDB<BlakeTwo256>, BlakeTwo256> {
+		let (mdb, root) = test_db(hashed_value);
 		TrieBackend::new(mdb, root)
 	}

 	#[test]
 	fn read_from_storage_returns_some() {
-		assert_eq!(test_trie().storage(b"key").unwrap(), Some(b"value".to_vec()));
+		read_from_storage_returns_some_inner(StateVersion::V0);
+		read_from_storage_returns_some_inner(StateVersion::V1);
+	}
+	fn read_from_storage_returns_some_inner(state_version: StateVersion) {
+		assert_eq!(test_trie(state_version).storage(b"key").unwrap(), Some(b"value".to_vec()));
 	}

 	#[test]
 	fn read_from_child_storage_returns_some() {
-		let test_trie = test_trie();
+		read_from_child_storage_returns_some_inner(StateVersion::V0);
+		read_from_child_storage_returns_some_inner(StateVersion::V1);
+	}
+	fn read_from_child_storage_returns_some_inner(state_version: StateVersion) {
+		let test_trie = test_trie(state_version);
 		assert_eq!(
 			test_trie
 				.child_storage(&ChildInfo::new_default(CHILD_KEY_1), b"value3")
@@ -341,12 +394,20 @@ pub mod tests {

 	#[test]
 	fn read_from_storage_returns_none() {
-		assert_eq!(test_trie().storage(b"non-existing-key").unwrap(), None);
+		read_from_storage_returns_none_inner(StateVersion::V0);
+		read_from_storage_returns_none_inner(StateVersion::V1);
+	}
+	fn read_from_storage_returns_none_inner(state_version: StateVersion) {
+		assert_eq!(test_trie(state_version).storage(b"non-existing-key").unwrap(), None);
 	}

 	#[test]
 	fn pairs_are_not_empty_on_non_empty_storage() {
-		assert!(!test_trie().pairs().is_empty());
+		pairs_are_not_empty_on_non_empty_storage_inner(StateVersion::V0);
+		pairs_are_not_empty_on_non_empty_storage_inner(StateVersion::V1);
+	}
+	fn pairs_are_not_empty_on_non_empty_storage_inner(state_version: StateVersion) {
+		assert!(!test_trie(state_version).pairs().is_empty());
 	}

 	#[test]
@@ -361,25 +422,35 @@ pub mod tests {

 	#[test]
 	fn storage_root_is_non_default() {
-		assert!(test_trie().storage_root(iter::empty()).0 != H256::repeat_byte(0));
+		storage_root_is_non_default_inner(StateVersion::V0);
+		storage_root_is_non_default_inner(StateVersion::V1);
 	}
-
-	#[test]
-	fn storage_root_transaction_is_empty() {
-		assert!(test_trie().storage_root(iter::empty()).1.drain().is_empty());
+	fn storage_root_is_non_default_inner(state_version: StateVersion) {
+		assert!(
+			test_trie(state_version).storage_root(iter::empty(), state_version).0 !=
+				H256::repeat_byte(0)
+		);
 	}

 	#[test]
 	fn storage_root_transaction_is_non_empty() {
-		let (new_root, mut tx) =
-			test_trie().storage_root(iter::once((&b"new-key"[..], Some(&b"new-value"[..]))));
+		storage_root_transaction_is_non_empty_inner(StateVersion::V0);
+		storage_root_transaction_is_non_empty_inner(StateVersion::V1);
+	}
+	fn storage_root_transaction_is_non_empty_inner(state_version: StateVersion) {
+		let (new_root, mut tx) = test_trie(state_version)
+			.storage_root(iter::once((&b"new-key"[..], Some(&b"new-value"[..]))), state_version);
 		assert!(!tx.drain().is_empty());
-		assert!(new_root != test_trie().storage_root(iter::empty()).0);
+		assert!(new_root != test_trie(state_version).storage_root(iter::empty(), state_version).0);
 	}

 	#[test]
 	fn prefix_walking_works() {
-		let trie = test_trie();
+		prefix_walking_works_inner(StateVersion::V0);
+		prefix_walking_works_inner(StateVersion::V1);
+	}
+	fn prefix_walking_works_inner(state_version: StateVersion) {
+		let trie = test_trie(state_version);

 		let mut seen = HashSet::new();
 		trie.for_keys_with_prefix(b"value", |key| {
@@ -20,20 +20,23 @@

 use crate::{backend::Consolidate, debug, warn, StorageKey, StorageValue};
 use codec::Encode;
-use hash_db::{self, Hasher, Prefix};
+use hash_db::{self, AsHashDB, HashDB, HashDBRef, Hasher, Prefix};
 #[cfg(feature = "std")]
 use parking_lot::RwLock;
 use sp_core::storage::ChildInfo;
 use sp_std::{boxed::Box, vec::Vec};
 use sp_trie::{
 	empty_child_trie_root, read_child_trie_value, read_trie_value,
-	trie_types::{Layout, TrieDB, TrieError},
-	DBValue, KeySpacedDB, MemoryDB, PrefixedMemoryDB, Trie, TrieDBIterator,
+	trie_types::{TrieDB, TrieError},
+	DBValue, KeySpacedDB, PrefixedMemoryDB, Trie, TrieDBIterator, TrieDBKeyIterator,
 };
 #[cfg(feature = "std")]
 use std::collections::HashMap;
 #[cfg(feature = "std")]
 use std::sync::Arc;
+// In this module, we only use layout for read operation and empty root,
+// where V1 and V0 are equivalent.
+use sp_trie::LayoutV1 as Layout;

 #[cfg(not(feature = "std"))]
 macro_rules! format {
@@ -178,7 +181,7 @@ where
 		child_info: Option<&ChildInfo>,
 		key: &[u8],
 	) -> Result<Option<StorageKey>> {
-		let dyn_eph: &dyn hash_db::HashDBRef<_, _>;
+		let dyn_eph: &dyn HashDBRef<_, _>;
 		let keyspace_eph;
 		if let Some(child_info) = child_info.as_ref() {
 			keyspace_eph = KeySpacedDB::new(self, child_info.keyspace());
@@ -189,7 +192,7 @@ where

 		let trie =
 			TrieDB::<H>::new(dyn_eph, root).map_err(|e| format!("TrieDB creation error: {}", e))?;
-		let mut iter = trie.iter().map_err(|e| format!("TrieDB iteration error: {}", e))?;
+		let mut iter = trie.key_iter().map_err(|e| format!("TrieDB iteration error: {}", e))?;

 		// The key just after the one given in input, basically `key++0`.
 		// Note: We are sure this is the next key if:
@@ -205,7 +208,7 @@ where
 		let next_element = iter.next();

 		let next_key = if let Some(next_element) = next_element {
-			let (next_key, _) =
+			let next_key =
 				next_element.map_err(|e| format!("TrieDB iterator next error: {}", e))?;
 			Some(next_key)
 		} else {
@@ -291,17 +294,7 @@ where
 			&self.root
 		};

-		let _ = self.trie_iter_inner(
-			root,
-			prefix,
-			|k, _v| {
-				f(&k);
-				true
-			},
-			child_info,
-			None,
-			false,
-		);
+		self.trie_iter_key_inner(root, prefix, |k| f(k), child_info)
 	}

 	/// Execute given closure for all keys starting with prefix.
@@ -320,32 +313,70 @@ where
 		};
 		let mut root = H::Out::default();
 		root.as_mut().copy_from_slice(&root_vec);
-		let _ = self.trie_iter_inner(
+		self.trie_iter_key_inner(
 			&root,
 			Some(prefix),
-			|k, _v| {
-				f(&k);
+			|k| {
+				f(k);
 				true
 			},
 			Some(child_info),
-			None,
-			false,
-		);
+		)
 	}

 	/// Execute given closure for all keys starting with prefix.
 	pub fn for_keys_with_prefix<F: FnMut(&[u8])>(&self, prefix: &[u8], mut f: F) {
-		let _ = self.trie_iter_inner(
+		self.trie_iter_key_inner(
 			&self.root,
 			Some(prefix),
-			|k, _v| {
-				f(&k);
+			|k| {
+				f(k);
 				true
 			},
 			None,
-			None,
-			false,
-		);
+		)
+	}
+
+	fn trie_iter_key_inner<F: FnMut(&[u8]) -> bool>(
+		&self,
+		root: &H::Out,
+		prefix: Option<&[u8]>,
+		mut f: F,
+		child_info: Option<&ChildInfo>,
+	) {
+		let mut iter = move |db| -> sp_std::result::Result<(), Box<TrieError<H::Out>>> {
+			let trie = TrieDB::<H>::new(db, root)?;
+			let iter = if let Some(prefix) = prefix.as_ref() {
+				TrieDBKeyIterator::new_prefixed(&trie, prefix)?
+			} else {
+				TrieDBKeyIterator::new(&trie)?
+			};
+
+			for x in iter {
+				let key = x?;
+
+				debug_assert!(prefix
+					.as_ref()
+					.map(|prefix| key.starts_with(prefix))
+					.unwrap_or(true));
+
+				if !f(&key) {
+					break
+				}
+			}
+
+			Ok(())
+		};
+
+		let result = if let Some(child_info) = child_info {
+			let db = KeySpacedDB::new(self, child_info.keyspace());
+			iter(&db)
+		} else {
+			iter(self)
+		};
+		if let Err(e) = result {
+			debug!(target: "trie", "Error while iterating by prefix: {}", e);
+		}
 	}

 	fn trie_iter_inner<F: FnMut(Vec<u8>, Vec<u8>) -> bool>(
@@ -414,13 +445,13 @@ pub(crate) struct Ephemeral<'a, S: 'a + TrieBackendStorage<H>, H: 'a + Hasher> {
 	overlay: &'a mut S::Overlay,
 }

-impl<'a, S: 'a + TrieBackendStorage<H>, H: 'a + Hasher> hash_db::AsHashDB<H, DBValue>
+impl<'a, S: 'a + TrieBackendStorage<H>, H: 'a + Hasher> AsHashDB<H, DBValue>
 	for Ephemeral<'a, S, H>
 {
-	fn as_hash_db<'b>(&'b self) -> &'b (dyn hash_db::HashDB<H, DBValue> + 'b) {
+	fn as_hash_db<'b>(&'b self) -> &'b (dyn HashDB<H, DBValue> + 'b) {
 		self
 	}
-	fn as_hash_db_mut<'b>(&'b mut self) -> &'b mut (dyn hash_db::HashDB<H, DBValue> + 'b) {
+	fn as_hash_db_mut<'b>(&'b mut self) -> &'b mut (dyn HashDB<H, DBValue> + 'b) {
 		self
 	}
 }
@@ -435,7 +466,7 @@ impl<'a, S: 'a + TrieBackendStorage<H>, H: Hasher> hash_db::HashDB<H, DBValue>
 	for Ephemeral<'a, S, H>
 {
 	fn get(&self, key: &H::Out, prefix: Prefix) -> Option<DBValue> {
-		if let Some(val) = hash_db::HashDB::get(self.overlay, key, prefix) {
+		if let Some(val) = HashDB::get(self.overlay, key, prefix) {
 			Some(val)
 		} else {
 			match self.storage.get(&key, prefix) {
@@ -449,38 +480,37 @@ impl<'a, S: 'a + TrieBackendStorage<H>, H: Hasher> hash_db::HashDB<H, DBValue>
 	}

 	fn contains(&self, key: &H::Out, prefix: Prefix) -> bool {
-		hash_db::HashDB::get(self, key, prefix).is_some()
+		HashDB::get(self, key, prefix).is_some()
 	}

 	fn insert(&mut self, prefix: Prefix, value: &[u8]) -> H::Out {
-		hash_db::HashDB::insert(self.overlay, prefix, value)
+		HashDB::insert(self.overlay, prefix, value)
 	}

 	fn emplace(&mut self, key: H::Out, prefix: Prefix, value: DBValue) {
-		hash_db::HashDB::emplace(self.overlay, key, prefix, value)
+		HashDB::emplace(self.overlay, key, prefix, value)
 	}

 	fn remove(&mut self, key: &H::Out, prefix: Prefix) {
-		hash_db::HashDB::remove(self.overlay, key, prefix)
+		HashDB::remove(self.overlay, key, prefix)
 	}
 }

-impl<'a, S: 'a + TrieBackendStorage<H>, H: Hasher> hash_db::HashDBRef<H, DBValue>
-	for Ephemeral<'a, S, H>
-{
+impl<'a, S: 'a + TrieBackendStorage<H>, H: Hasher> HashDBRef<H, DBValue> for Ephemeral<'a, S, H> {
 	fn get(&self, key: &H::Out, prefix: Prefix) -> Option<DBValue> {
-		hash_db::HashDB::get(self, key, prefix)
+		HashDB::get(self, key, prefix)
 	}

 	fn contains(&self, key: &H::Out, prefix: Prefix) -> bool {
-		hash_db::HashDB::contains(self, key, prefix)
+		HashDB::contains(self, key, prefix)
 	}
 }

 /// Key-value pairs storage that is used by trie backend essence.
 pub trait TrieBackendStorage<H: Hasher>: Send + Sync {
 	/// Type of in-memory overlay.
-	type Overlay: hash_db::HashDB<H, DBValue> + Default + Consolidate;
+	type Overlay: HashDB<H, DBValue> + Default + Consolidate;
+
 	/// Get the value stored at key.
 	fn get(&self, key: &H::Out, prefix: Prefix) -> Result<Option<DBValue>>;
 }
@@ -495,35 +525,28 @@ impl<H: Hasher> TrieBackendStorage<H> for Arc<dyn Storage<H>> {
 	}
 }

-// This implementation is used by test storage trie clients.
-impl<H: Hasher> TrieBackendStorage<H> for PrefixedMemoryDB<H> {
-	type Overlay = PrefixedMemoryDB<H>;
-
-	fn get(&self, key: &H::Out, prefix: Prefix) -> Result<Option<DBValue>> {
-		Ok(hash_db::HashDB::get(self, key, prefix))
-	}
-}
-
-impl<H: Hasher> TrieBackendStorage<H> for MemoryDB<H> {
-	type Overlay = MemoryDB<H>;
-
-	fn get(&self, key: &H::Out, prefix: Prefix) -> Result<Option<DBValue>> {
-		Ok(hash_db::HashDB::get(self, key, prefix))
-	}
-}
-
-impl<S: TrieBackendStorage<H>, H: Hasher> hash_db::AsHashDB<H, DBValue>
-	for TrieBackendEssence<S, H>
+impl<H, KF> TrieBackendStorage<H> for sp_trie::GenericMemoryDB<H, KF>
+where
+	H: Hasher,
+	KF: sp_trie::KeyFunction<H> + Send + Sync,
 {
-	fn as_hash_db<'b>(&'b self) -> &'b (dyn hash_db::HashDB<H, DBValue> + 'b) {
+	type Overlay = Self;
+
+	fn get(&self, key: &H::Out, prefix: Prefix) -> Result<Option<DBValue>> {
+		Ok(hash_db::HashDB::get(self, key, prefix))
+	}
+}
+
+impl<S: TrieBackendStorage<H>, H: Hasher> AsHashDB<H, DBValue> for TrieBackendEssence<S, H> {
+	fn as_hash_db<'b>(&'b self) -> &'b (dyn HashDB<H, DBValue> + 'b) {
 		self
 	}
-	fn as_hash_db_mut<'b>(&'b mut self) -> &'b mut (dyn hash_db::HashDB<H, DBValue> + 'b) {
+	fn as_hash_db_mut<'b>(&'b mut self) -> &'b mut (dyn HashDB<H, DBValue> + 'b) {
 		self
 	}
 }

-impl<S: TrieBackendStorage<H>, H: Hasher> hash_db::HashDB<H, DBValue> for TrieBackendEssence<S, H> {
+impl<S: TrieBackendStorage<H>, H: Hasher> HashDB<H, DBValue> for TrieBackendEssence<S, H> {
 	fn get(&self, key: &H::Out, prefix: Prefix) -> Option<DBValue> {
 		if *key == self.empty {
 			return Some([0u8].to_vec())
@@ -538,7 +561,7 @@ impl<S: TrieBackendStorage<H>, H: Hasher> hash_db::HashDB<H, DBValue> for TrieBa
 	}

 	fn contains(&self, key: &H::Out, prefix: Prefix) -> bool {
-		hash_db::HashDB::get(self, key, prefix).is_some()
+		HashDB::get(self, key, prefix).is_some()
 	}

 	fn insert(&mut self, _prefix: Prefix, _value: &[u8]) -> H::Out {
@@ -554,15 +577,13 @@ impl<S: TrieBackendStorage<H>, H: Hasher> hash_db::HashDB<H, DBValue> for TrieBa
 	}
 }

-impl<S: TrieBackendStorage<H>, H: Hasher> hash_db::HashDBRef<H, DBValue>
-	for TrieBackendEssence<S, H>
-{
+impl<S: TrieBackendStorage<H>, H: Hasher> HashDBRef<H, DBValue> for TrieBackendEssence<S, H> {
 	fn get(&self, key: &H::Out, prefix: Prefix) -> Option<DBValue> {
-		hash_db::HashDB::get(self, key, prefix)
+		HashDB::get(self, key, prefix)
 	}

 	fn contains(&self, key: &H::Out, prefix: Prefix) -> bool {
-		hash_db::HashDB::contains(self, key, prefix)
+		HashDB::contains(self, key, prefix)
 	}
 }

@@ -570,7 +591,9 @@ impl<S: TrieBackendStorage<H>, H: Hasher> hash_db::HashDBRef<H, DBValue>
 mod test {
 	use super::*;
 	use sp_core::{Blake2Hasher, H256};
-	use sp_trie::{trie_types::TrieDBMut, KeySpacedDBMut, PrefixedMemoryDB, TrieMut};
+	use sp_trie::{
+		trie_types::TrieDBMutV1 as TrieDBMut, KeySpacedDBMut, PrefixedMemoryDB, TrieMut,
+	};

 	#[test]
 	fn next_storage_key_and_next_child_storage_key_work() {