diff --git a/storage/diemdb/src/lib.rs b/storage/diemdb/src/lib.rs
index 754f3a71016a3..3e81bc8282655 100644
--- a/storage/diemdb/src/lib.rs
+++ b/storage/diemdb/src/lib.rs
@@ -233,6 +233,7 @@ impl DiemDB {
             TRANSACTION_CF_NAME,
             TRANSACTION_ACCUMULATOR_CF_NAME,
             TRANSACTION_BY_ACCOUNT_CF_NAME,
+            TRANSACTION_BY_HASH_CF_NAME,
             TRANSACTION_INFO_CF_NAME,
         ]
     }
diff --git a/storage/diemdb/src/schema/mod.rs b/storage/diemdb/src/schema/mod.rs
index 5f218e37cedad..fcef9d0d254ee 100644
--- a/storage/diemdb/src/schema/mod.rs
+++ b/storage/diemdb/src/schema/mod.rs
@@ -18,6 +18,7 @@ pub(crate) mod stale_node_index;
 pub(crate) mod transaction;
 pub(crate) mod transaction_accumulator;
 pub(crate) mod transaction_by_account;
+pub(crate) mod transaction_by_hash;
 pub(crate) mod transaction_info;
 
 use anyhow::{ensure, Result};
@@ -34,6 +35,7 @@ pub const STALE_NODE_INDEX_CF_NAME: ColumnFamilyName = "stale_node_index";
 pub const TRANSACTION_CF_NAME: ColumnFamilyName = "transaction";
 pub const TRANSACTION_ACCUMULATOR_CF_NAME: ColumnFamilyName = "transaction_accumulator";
 pub const TRANSACTION_BY_ACCOUNT_CF_NAME: ColumnFamilyName = "transaction_by_account";
+pub const TRANSACTION_BY_HASH_CF_NAME: ColumnFamilyName = "transaction_by_hash";
 pub const TRANSACTION_INFO_CF_NAME: ColumnFamilyName = "transaction_info";
 
 fn ensure_slice_len_eq(data: &[u8], len: usize) -> Result<()> {
@@ -91,6 +93,7 @@ pub mod fuzzing {
                 super::transaction_by_account::TransactionByAccountSchema,
                 data
             );
+            decode_key_value!(super::transaction_by_hash::TransactionByHashSchema, data);
             decode_key_value!(super::transaction_info::TransactionInfoSchema, data);
         }
     }
diff --git a/storage/diemdb/src/schema/transaction_by_hash/mod.rs b/storage/diemdb/src/schema/transaction_by_hash/mod.rs
new file mode 100644
index 0000000000000..25234f73c8928
--- /dev/null
+++ b/storage/diemdb/src/schema/transaction_by_hash/mod.rs
@@ -0,0 +1,54 @@
+// Copyright (c) The Diem Core Contributors
+// SPDX-License-Identifier: Apache-2.0
+
+//! This module defines physical storage schema mapping transaction hash to its version.
+//! With the version one can resort to `TransactionSchema` for the transaction content.
+//!
+//! ```text
+//! |<--key-->|<-value->|
+//! |   hash  | txn_ver |
+//! ```
+
+use crate::schema::{ensure_slice_len_eq, TRANSACTION_BY_HASH_CF_NAME};
+use anyhow::Result;
+use byteorder::{BigEndian, ReadBytesExt};
+use diem_crypto::HashValue;
+use diem_types::transaction::Version;
+use schemadb::{
+    define_schema,
+    schema::{KeyCodec, ValueCodec},
+};
+use std::mem::size_of;
+
+define_schema!(
+    TransactionByHashSchema,
+    HashValue,
+    Version,
+    TRANSACTION_BY_HASH_CF_NAME
+);
+
+impl KeyCodec<TransactionByHashSchema> for HashValue {
+    fn encode_key(&self) -> Result<Vec<u8>> {
+        Ok(self.to_vec())
+    }
+
+    fn decode_key(data: &[u8]) -> Result<Self> {
+        ensure_slice_len_eq(data, size_of::<Self>())?;
+        Ok(HashValue::from_slice(data)?)
+    }
+}
+
+impl ValueCodec<TransactionByHashSchema> for Version {
+    fn encode_value(&self) -> Result<Vec<u8>> {
+        Ok(self.to_be_bytes().to_vec())
+    }
+
+    fn decode_value(mut data: &[u8]) -> Result<Self> {
+        ensure_slice_len_eq(data, size_of::<Self>())?;
+
+        Ok(data.read_u64::<BigEndian>()?)
+    }
+}
+
+#[cfg(test)]
+mod test;
diff --git a/storage/diemdb/src/schema/transaction_by_hash/test.rs b/storage/diemdb/src/schema/transaction_by_hash/test.rs
new file mode 100644
index 0000000000000..e802ef0c49f7d
--- /dev/null
+++ b/storage/diemdb/src/schema/transaction_by_hash/test.rs
@@ -0,0 +1,16 @@
+// Copyright (c) The Diem Core Contributors
+// SPDX-License-Identifier: Apache-2.0
+
+use super::*;
+use proptest::prelude::*;
+use schemadb::schema::assert_encode_decode;
+
+proptest! {
+    #[test]
+    fn test_encode_decode(
+        hash in any::<HashValue>(),
+        version in any::<Version>(),
+    ) {
+        assert_encode_decode::<TransactionByHashSchema>(&hash, &version);
+    }
+}