diff --git a/production/db/core/inc/db_internal_types.hpp b/production/db/core/inc/db_internal_types.hpp
index ec66345bcaec..24489f0d26e2 100644
--- a/production/db/core/inc/db_internal_types.hpp
+++ b/production/db/core/inc/db_internal_types.hpp
@@ -19,7 +19,6 @@
 #include "gaia_internal/db/db_object.hpp"
 #include "gaia_internal/db/db_types.hpp"
 
-#include "base_index.hpp"
 #include "memory_types.hpp"
 
 namespace gaia
@@ -197,7 +196,7 @@ struct id_index_t
 // These are types meant to access index types from the client/server.
 namespace index
 {
-
+class base_index_t;
 typedef std::shared_ptr<base_index_t> db_index_t;
 typedef std::unordered_map<common::gaia_id_t, db_index_t> indexes_t;
 
diff --git a/production/db/core/src/db_server.cpp b/production/db/core/src/db_server.cpp
index 99fbe6b966be..81377275ef26 100644
--- a/production/db/core/src/db_server.cpp
+++ b/production/db/core/src/db_server.cpp
@@ -751,7 +751,7 @@ void server_t::init_indexes()
 
         for (const auto& index : catalog_core_t::list_indexes(type_record_id))
         {
-            index::index_builder_t::populate_index(index.id(), obj->type, locator);
+            index::index_builder_t::populate_index(index.id(), locator);
         }
     }
 }
diff --git a/production/db/core/src/index_builder.cpp b/production/db/core/src/index_builder.cpp
index 70cdd82a7c77..2e8c23c8ba14 100644
--- a/production/db/core/src/index_builder.cpp
+++ b/production/db/core/src/index_builder.cpp
@@ -5,8 +5,9 @@
 
 #include "gaia_internal/db/index_builder.hpp"
 
-#include <unordered_set>
+#include <array>
 #include <utility>
+#include <vector>
 
 #include "gaia/exceptions.hpp"
 
@@ -30,29 +31,11 @@ namespace db
 namespace index
 {
 
-index_key_t index_builder_t::make_key(gaia_id_t index_id, gaia_type_t type_id, const uint8_t* payload)
+index_key_t index_builder_t::make_key(db_index_t index, const uint8_t* payload)
 {
     ASSERT_PRECONDITION(payload, "Cannot compute key on null payloads.");
 
-    index_key_t index_key;
-    gaia_id_t type_record_id = type_id_mapping_t::instance().get_record_id(type_id);
-
-    ASSERT_INVARIANT(
-        type_record_id != c_invalid_gaia_id,
-        "The type '" + std::to_string(type_id) + "' does not exist in the catalog.");
-
-    auto table = catalog_core_t::get_table(type_record_id);
-    auto schema = table.binary_schema();
-    auto index_view = index_view_t(id_to_ptr(index_id));
-
-    const auto& fields = *(index_view.fields());
-    for (gaia_id_t field_id : fields)
-    {
-        field_position_t pos = field_view_t(id_to_ptr(field_id)).position();
-        index_key.insert(payload_types::get_field_value(type_id, payload, schema->data(), schema->size(), pos));
-    }
-
-    return index_key;
+    return index_key_t(index->key_schema(), payload);
 }
 
 void index_builder_t::serialize_key(const index_key_t& key, payload_types::data_write_buffer_t& buffer)
@@ -106,12 +89,21 @@ indexes_t::iterator index_builder_t::create_empty_index(const index_view_t& inde
     ASSERT_PRECONDITION(skip_catalog_integrity_check || index_view.table_id() != c_invalid_gaia_id, "Cannot find table for index.");
 
     bool is_unique = index_view.unique();
-    gaia_type_t table_type = c_invalid_gaia_type;
+
+    index_key_schema_t key_schema;
 
     if (index_view.table_id() != c_invalid_gaia_id)
     {
         auto table_view = table_view_t(id_to_ptr(index_view.table_id()));
-        table_type = table_view.table_type();
+        key_schema.table_type = table_view.table_type();
+        key_schema.binary_schema = table_view.binary_schema();
+
+        const auto& fields = *(index_view.fields());
+        for (gaia_id_t field_id : fields)
+        {
+            field_position_t pos = field_view_t(id_to_ptr(field_id)).position();
+            key_schema.field_positions.push_back(pos);
+        }
     }
 
     switch (index_view.type())
@@ -119,9 +111,9 @@ indexes_t::iterator index_builder_t::create_empty_index(const index_view_t& inde
     case catalog::index_type_t::range:
         return get_indexes()->emplace(
                                 index_view.id(),
-                                std::make_shared<range_index_t, gaia_id_t, gaia_type_t, bool>(
+                                std::make_shared<range_index_t, gaia_id_t, index_key_schema_t, bool>(
                                     std::forward<gaia_id_t>(index_view.id()),
-                                    std::forward<gaia_type_t>(table_type),
+                                    std::forward<index_key_schema_t>(key_schema),
                                     std::forward<bool>(is_unique)))
             .first;
         break;
@@ -129,9 +121,9 @@ indexes_t::iterator index_builder_t::create_empty_index(const index_view_t& inde
     case catalog::index_type_t::hash:
         return get_indexes()->emplace(
                                 index_view.id(),
-                                std::make_shared<hash_index_t, gaia_id_t, gaia_type_t, bool>(
+                                std::make_shared<hash_index_t, gaia_id_t, index_key_schema_t, bool>(
                                     std::forward<gaia_id_t>(index_view.id()),
-                                    std::forward<gaia_type_t>(table_type),
+                                    std::forward<index_key_schema_t>(key_schema),
                                     std::forward<bool>(is_unique)))
             .first;
         break;
@@ -254,42 +246,24 @@ void update_index_entry(
 }
 
 void index_builder_t::update_index(
-    gaia_id_t index_id, index_key_t&& key, index_record_t record, bool allow_create_empty)
+    db_index_t index, index_key_t&& key, index_record_t record)
 {
-    ASSERT_PRECONDITION(get_indexes(), "Indexes are not initialized.");
-
-    auto it = get_indexes()->find(index_id);
-
-    if (allow_create_empty && it == get_indexes()->end())
-    {
-        auto index_ptr = id_to_ptr(index_id);
-        ASSERT_INVARIANT(index_ptr != nullptr, "Cannot find index in catalog.");
-        auto index_view = index_view_t(index_ptr);
-        it = index::index_builder_t::create_empty_index(index_view);
-    }
-    else
-    {
-        ASSERT_INVARIANT(it != get_indexes()->end(), "Index structure could not be found.");
-    }
+    bool is_unique_index = index->is_unique();
 
-    bool is_unique_index = it->second->is_unique();
-
-    switch (it->second->type())
+    switch (index->type())
     {
     case catalog::index_type_t::range:
-        update_index_entry<range_index_t>(it->second.get(), is_unique_index, std::move(key), record);
+        update_index_entry<range_index_t>(index.get(), is_unique_index, std::move(key), record);
         break;
     case catalog::index_type_t::hash:
-        update_index_entry<hash_index_t>(it->second.get(), is_unique_index, std::move(key), record);
+        update_index_entry<hash_index_t>(index.get(), is_unique_index, std::move(key), record);
         break;
     }
 }
 
 void index_builder_t::update_index(
-    gaia::common::gaia_id_t index_id,
-    gaia_type_t type_id,
-    const txn_log_t::log_record_t& log_record,
-    bool allow_create_empty)
+    db_index_t index,
+    const txn_log_t::log_record_t& log_record)
 {
     // Most operations expect an object located at new_offset,
     // so we'll try to get a reference to its payload.
@@ -300,18 +274,17 @@ void index_builder_t::update_index(
     {
     case gaia_operation_t::create:
         index_builder_t::update_index(
-            index_id,
-            index_builder_t::make_key(index_id, type_id, payload),
+            index,
+            index_builder_t::make_key(index, payload),
             index_builder_t::make_record(
-                log_record.locator, log_record.new_offset, index_record_operation_t::insert),
-            allow_create_empty);
+                log_record.locator, log_record.new_offset, index_record_operation_t::insert));
         break;
     case gaia_operation_t::update:
     {
         auto old_obj = offset_to_ptr(log_record.old_offset);
         auto old_payload = (old_obj) ? reinterpret_cast<const uint8_t*>(old_obj->data()) : nullptr;
-        index_key_t old_key = index_builder_t::make_key(index_id, type_id, old_payload);
-        index_key_t new_key = index_builder_t::make_key(index_id, type_id, payload);
+        index_key_t old_key = index_builder_t::make_key(index, old_payload);
+        index_key_t new_key = index_builder_t::make_key(index, payload);
 
         // If the index key is not changed, mark operation as an update.
         // Otherwise, we'll mark it as two individual remove/insert operations,
@@ -319,26 +292,23 @@ void index_builder_t::update_index(
         if (new_key == old_key)
         {
             index_builder_t::update_index(
-                index_id,
+                index,
                 std::move(new_key),
                 index_builder_t::make_record(
-                    log_record.locator, log_record.new_offset, index_record_operation_t::update),
-                allow_create_empty);
+                    log_record.locator, log_record.new_offset, index_record_operation_t::update));
         }
         else
         {
             index_builder_t::update_index(
-                index_id,
+                index,
                 std::move(old_key),
                 index_builder_t::make_record(
-                    log_record.locator, log_record.old_offset, index_record_operation_t::remove),
-                allow_create_empty);
+                    log_record.locator, log_record.old_offset, index_record_operation_t::remove));
             index_builder_t::update_index(
-                index_id,
+                index,
                 std::move(new_key),
                 index_builder_t::make_record(
-                    log_record.locator, log_record.new_offset, index_record_operation_t::insert),
-                allow_create_empty);
+                    log_record.locator, log_record.new_offset, index_record_operation_t::insert));
         }
     }
     break;
@@ -347,11 +317,10 @@ void index_builder_t::update_index(
         auto old_obj = offset_to_ptr(log_record.old_offset);
         auto old_payload = (old_obj) ? reinterpret_cast<const uint8_t*>(old_obj->data()) : nullptr;
         index_builder_t::update_index(
-            index_id,
-            index_builder_t::make_key(index_id, type_id, old_payload),
+            index,
+            index_builder_t::make_key(index, old_payload),
             index_builder_t::make_record(
-                log_record.locator, log_record.old_offset, index_record_operation_t::remove),
-            allow_create_empty);
+                log_record.locator, log_record.old_offset, index_record_operation_t::remove));
     }
     break;
     default:
@@ -360,12 +329,18 @@ void index_builder_t::update_index(
     }
 }
 
-void index_builder_t::populate_index(common::gaia_id_t index_id, common::gaia_type_t type_id, gaia_locator_t locator)
+void index_builder_t::populate_index(common::gaia_id_t index_id, gaia_locator_t locator)
 {
+    ASSERT_PRECONDITION(get_indexes(), "Indexes are not initialized.");
     auto payload = reinterpret_cast<const uint8_t*>(locator_to_ptr(locator)->data());
+
+    auto it = get_indexes()->find(index_id);
+    ASSERT_INVARIANT(it != get_indexes()->end(), "Index structure could not be found.");
+    db_index_t index = it->second;
+
     update_index(
-        index_id,
-        make_key(index_id, type_id, payload),
+        index,
+        make_key(index, payload),
         make_record(locator, locator_to_offset(locator), index_record_operation_t::insert));
 }
 
@@ -381,7 +356,7 @@ void index_builder_t::update_indexes_from_txn_log(
     // table is created or dropped in the txn.
     // Keep track of dropped tables.
     bool has_cleared_cache = false;
-    std::unordered_set<gaia_type_t> dropped_types;
+    std::vector<gaia_type_t> dropped_types;
 
     for (size_t i = 0; i < records.record_count; ++i)
     {
@@ -403,7 +378,7 @@ void index_builder_t::update_indexes_from_txn_log(
             if (log_record.operation == gaia_operation_t::remove)
             {
                 auto table_view = table_view_t(offset_to_ptr(log_record.old_offset));
-                dropped_types.insert(table_view.table_type());
+                dropped_types.push_back(table_view.table_type());
             }
         }
     }
@@ -448,7 +423,7 @@ void index_builder_t::update_indexes_from_txn_log(
         // The operation is from a dropped table.
         // Skip if catalog verification disabled and type not found in the catalog.
         if (is_system_object(obj->type)
-            || dropped_types.find(obj->type) != dropped_types.end()
+            || std::find(dropped_types.begin(), dropped_types.end(), obj->type) != dropped_types.end()
             || (skip_catalog_integrity_check && type_record_id == c_invalid_gaia_id))
         {
             continue;
@@ -456,68 +431,94 @@ void index_builder_t::update_indexes_from_txn_log(
 
         for (const auto& index : catalog_core_t::list_indexes(type_record_id))
         {
-            index::index_builder_t::update_index(index.id(), obj->type, log_record, allow_create_empty);
+            ASSERT_PRECONDITION(get_indexes(), "Indexes are not initialized.");
+            auto it = get_indexes()->find(index.id());
+
+            if (allow_create_empty && it == get_indexes()->end())
+            {
+                auto index_ptr = id_to_ptr(index.id());
+                ASSERT_INVARIANT(index_ptr != nullptr, "Cannot find index in catalog.");
+                auto index_view = index_view_t(index_ptr);
+                it = index::index_builder_t::create_empty_index(index_view);
+            }
+            else
+            {
+                ASSERT_INVARIANT(it != get_indexes()->end(), "Index structure could not be found.");
+            }
+
+            index::index_builder_t::update_index(it->second, log_record);
         }
     }
 }
 
 template <class T_index>
-void remove_entries_with_offsets(base_index_t* base_index, const std::unordered_set<gaia_offset_t>& offsets, gaia_txn_id_t txn_id)
+void remove_entries_with_offsets(base_index_t* base_index, const index_offset_buffer_t& offsets, gaia_txn_id_t txn_id)
 {
     auto index = static_cast<T_index*>(base_index);
-    index->remove_index_entry_with_offsets(offsets, txn_id);
-}
 
-void index_builder_t::gc_indexes_from_txn_log(const txn_log_t& records, bool deallocate_new_offsets)
-{
-    std::unordered_set<gaia_offset_t> collected_offsets;
-    std::unordered_set<gaia_type_t> offset_types;
-
-    for (size_t i = 0; i < records.record_count; ++i)
+    for (size_t i = 0; i < offsets.size(); ++i)
     {
-        const auto& log_record = records.log_records[i];
-        gaia_offset_t offset = deallocate_new_offsets ? log_record.new_offset : log_record.old_offset;
-
-        // If no action is needed, move on to the next log record.
-        if (offset != c_invalid_gaia_offset)
+        if (offsets.get_type(i) == index->table_type())
         {
+            gaia_offset_t offset = offsets.get_offset(i);
             auto obj = offset_to_ptr(offset);
+            index_key_t key = index_key_t(index->key_schema(), reinterpret_cast<const uint8_t*>(obj->data()));
 
-            // We do not index system objects, so we can move on.
-            if (is_system_object(obj->type))
-            {
-                continue;
-            }
-
-            collected_offsets.insert(offset);
-            offset_types.insert(obj->type);
+            index->remove_index_entry_with_offset(key, offset, txn_id);
         }
     }
+}
 
-    // Nothing to do here.
-    if (collected_offsets.size() == 0)
+void index_builder_t::gc_indexes_from_txn_log(const txn_log_t& records, bool deallocate_new_offsets)
+{
+    size_t records_index = 0;
+    while (records_index < records.record_count)
     {
-        return;
-    }
+        index_offset_buffer_t collected_offsets;
+        // Fill the offset buffer for garbage collection.
+        // Exit the loop when we either have run out of records to process or the offsets buffer is full.
+        for (; records_index < records.record_count && collected_offsets.size() < c_offset_buffer_size; ++records_index)
+        {
+            const auto& log_record = records.log_records[records_index];
 
-    for (auto it : *get_indexes())
-    {
-        gaia_type_t indexed_type = it.second->table_type();
+            gaia_offset_t offset = deallocate_new_offsets ? log_record.new_offset : log_record.old_offset;
+
+            // If no action is needed, move on to the next log record.
+            if (offset != c_invalid_gaia_offset)
+            {
+                auto obj = offset_to_ptr(offset);
+
+                // We do not index system objects, so we can move on.
+                if (is_system_object(obj->type))
+                {
+                    continue;
+                }
+
+                // Add the offset to the buffers and advance the buffer index.
+                collected_offsets.insert(offset, obj->type);
+            }
+        }
 
-        // This index does not contain any of the deleted offsets.
-        if (offset_types.find(indexed_type) == offset_types.end())
+        // When we reach this point, either we have 1) run out of records to iterate over or 2) the offsets buffer is now considered full.
+        // We know that 2) is false when the offsets buffer is empty and there is no garbage to collect.
+        // Therefore we can safely return here.
+        if (collected_offsets.empty())
         {
-            continue;
+            return;
         }
 
-        switch (it.second->type())
+        // Garbage collect the offsets in the buffer.
+        for (const auto& it : *get_indexes())
         {
-        case catalog::index_type_t::range:
-            remove_entries_with_offsets<range_index_t>(it.second.get(), collected_offsets, records.begin_ts);
-            break;
-        case catalog::index_type_t::hash:
-            remove_entries_with_offsets<hash_index_t>(it.second.get(), collected_offsets, records.begin_ts);
-            break;
+            switch (it.second->type())
+            {
+            case catalog::index_type_t::range:
+                remove_entries_with_offsets<range_index_t>(it.second.get(), collected_offsets, records.begin_ts);
+                break;
+            case catalog::index_type_t::hash:
+                remove_entries_with_offsets<hash_index_t>(it.second.get(), collected_offsets, records.begin_ts);
+                break;
+            }
         }
     }
 }
@@ -531,7 +532,7 @@ void mark_index_entries(base_index_t* base_index, gaia_txn_id_t txn_id)
 
 void index_builder_t::mark_index_entries_committed(gaia_txn_id_t txn_id)
 {
-    for (auto it : *get_indexes())
+    for (const auto& it : *get_indexes())
     {
         // Optimization: only mark index entries committed for UNIQUE indexes, as we only look up the flags on that path.
         if (it.second->is_unique())
diff --git a/production/db/inc/index/base_index.hpp b/production/db/inc/index/base_index.hpp
index d760ab0a6e29..8380e670dea0 100644
--- a/production/db/inc/index/base_index.hpp
+++ b/production/db/inc/index/base_index.hpp
@@ -22,6 +22,8 @@
 #include "gaia_internal/common/generator_iterator.hpp"
 #include "gaia_internal/db/db_types.hpp"
 
+#include "index_key.hpp"
+
 namespace gaia
 {
 namespace db
@@ -82,13 +84,11 @@ constexpr void mark_committed(index_record_t& record)
     record.flags |= c_mark_committed_mask;
 }
 
-class index_key_t;
-
 class base_index_t
 {
 public:
-    base_index_t(gaia::common::gaia_id_t index_id, catalog::index_type_t index_type, common::gaia_type_t table_type, bool is_unique)
-        : m_index_id(index_id), m_index_type(index_type), m_table_type(table_type), m_is_unique(is_unique)
+    base_index_t(gaia::common::gaia_id_t index_id, catalog::index_type_t index_type, index_key_schema_t key_schema, bool is_unique)
+        : m_index_id(index_id), m_index_type(index_type), m_key_schema(key_schema), m_is_unique(is_unique)
     {
     }
 
@@ -97,6 +97,7 @@ class base_index_t
     gaia::common::gaia_id_t id() const;
     catalog::index_type_t type() const;
     gaia::common::gaia_type_t table_type() const;
+    const index_key_schema_t& key_schema() const;
     bool is_unique() const;
 
     std::recursive_mutex& get_lock() const;
@@ -110,7 +111,7 @@ class base_index_t
 protected:
     gaia::common::gaia_id_t m_index_id;
     catalog::index_type_t m_index_type;
-    gaia::common::gaia_type_t m_table_type;
+    index_key_schema_t m_key_schema;
     bool m_is_unique;
 
     // Recursive_mutex is used here because shared_mutex cannot be unlocked multiple times on the same thread.
diff --git a/production/db/inc/index/hash_index.hpp b/production/db/inc/index/hash_index.hpp
index b271dd8191a9..8a904e7a61dd 100644
--- a/production/db/inc/index/hash_index.hpp
+++ b/production/db/inc/index/hash_index.hpp
@@ -27,8 +27,8 @@ using hash_index_iterator_t = index_iterator_t<hash_type_t, hash_type_t::const_i
 class hash_index_t : public index_t<hash_type_t, hash_index_iterator_t>
 {
 public:
-    hash_index_t(gaia::common::gaia_id_t index_id, gaia::common::gaia_type_t table_type, bool is_unique = false)
-        : index_t(index_id, catalog::index_type_t::hash, table_type, is_unique)
+    hash_index_t(gaia::common::gaia_id_t index_id, index_key_schema_t key_schema, bool is_unique = false)
+        : index_t(index_id, catalog::index_type_t::hash, key_schema, is_unique)
     {
     }
     ~hash_index_t() = default;
diff --git a/production/db/inc/index/index.hpp b/production/db/inc/index/index.hpp
index fc39a4752d29..a17c3e9acb28 100644
--- a/production/db/inc/index/index.hpp
+++ b/production/db/inc/index/index.hpp
@@ -5,7 +5,8 @@
 
 #pragma once
 
-#include <unordered_set>
+#include <array>
+#include <utility>
 
 #include "gaia_internal/db/db_types.hpp"
 
@@ -48,6 +49,26 @@ class index_writer_guard_t
     T_structure& m_data;
 };
 
+constexpr size_t c_offset_buffer_size = 32;
+
+/*
+* Buffer storing data for garbage collecting offsets.
+*/
+
+class index_offset_buffer_t
+{
+public:
+    void insert(gaia_offset_t offset, common::gaia_type_t type);
+    gaia_offset_t get_offset(size_t index) const;
+    common::gaia_type_t get_type(size_t index) const;
+    bool empty() const;
+    size_t size() const;
+
+private:
+    std::array<std::pair<gaia_offset_t, common::gaia_type_t>, c_offset_buffer_size> m_offsets = {};
+    size_t m_size = 0;
+};
+
 /**
  * Abstract in-memory index type:
  * T_structure is the underlying backing data structure of the index.
@@ -57,8 +78,8 @@ template <typename T_structure, typename T_iterator>
 class index_t : public base_index_t
 {
 public:
-    index_t(gaia::common::gaia_id_t index_id, catalog::index_type_t index_type, common::gaia_type_t table_type, bool is_unique)
-        : base_index_t(index_id, index_type, table_type, is_unique)
+    index_t(gaia::common::gaia_id_t index_id, catalog::index_type_t index_type, index_key_schema_t key_schema, bool is_unique)
+        : base_index_t(index_id, index_type, key_schema, is_unique)
     {
     }
     ~index_t() override = default;
@@ -71,7 +92,7 @@ class index_t : public base_index_t
 
     // Index structure maintenance.
     void insert_index_entry(index_key_t&& key, index_record_t record);
-    void remove_index_entry_with_offsets(const std::unordered_set<gaia_offset_t>& offsets, gaia_txn_id_t gc_txn_id);
+    void remove_index_entry_with_offset(const index_key_t& key, gaia_offset_t offset, gaia_txn_id_t gc_txn_id);
 
     // This method will mark all entries below a specified txn_id as committed.
     // This must only be called after all aborted/terminated index entries below the txn_id are garbage collected.
diff --git a/production/db/inc/index/index.inc b/production/db/inc/index/index.inc
index 61baf85f4fd8..dd6a64ef0340 100644
--- a/production/db/inc/index/index.inc
+++ b/production/db/inc/index/index.inc
@@ -4,7 +4,6 @@
 /////////////////////////////////////////////
 
 // Index structure maintenance.
-
 template <typename T_structure, typename T_iterator>
 void index_t<T_structure, T_iterator>::insert_index_entry(index_key_t&& key, index_record_t record)
 {
@@ -29,7 +28,7 @@ void index_t<T_structure, T_iterator>::insert_index_entry(index_key_t&& key, ind
 }
 
 template <typename T_structure, typename T_iterator>
-void index_t<T_structure, T_iterator>::remove_index_entry_with_offsets(const std::unordered_set<gaia_offset_t>& offsets, gaia_txn_id_t gc_txn_id)
+void index_t<T_structure, T_iterator>::remove_index_entry_with_offset(const index_key_t& key, gaia_offset_t offset, gaia_txn_id_t gc_txn_id)
 {
     std::lock_guard lock(m_index_lock);
 
@@ -40,9 +39,11 @@ void index_t<T_structure, T_iterator>::remove_index_entry_with_offsets(const std
         return;
     }
 
-    for (auto it = m_data.begin(); it != m_data.end();)
+    auto range = m_data.equal_range(key);
+
+    for (auto& it = range.first; it != range.second;)
     {
-        if (offsets.find(it->second.offset) != offsets.end())
+        if (it->second.offset == offset)
         {
             it = m_data.erase(it);
         }
diff --git a/production/db/inc/index/index_key.hpp b/production/db/inc/index/index_key.hpp
index ab68e0b6b995..6e104453ee7f 100644
--- a/production/db/inc/index/index_key.hpp
+++ b/production/db/inc/index/index_key.hpp
@@ -5,9 +5,15 @@
 
 #pragma once
 
+#include <cstdint>
+
 #include <ostream>
 #include <vector>
 
+#include <flatbuffers/reflection.h>
+
+#include "gaia/common.hpp"
+
 #include "data_holder.hpp"
 
 namespace gaia
@@ -17,6 +23,17 @@ namespace db
 namespace index
 {
 
+/**
+ * Schema required to compute index keys from a binary payload.
+ **/
+
+struct index_key_schema_t
+{
+    common::gaia_type_t table_type;
+    const flatbuffers::Vector<uint8_t>* binary_schema;
+    common::field_position_list_t field_positions;
+};
+
 /**
 * Reflection based logical key for the index.
 */
@@ -27,6 +44,7 @@ class index_key_t
 
 public:
     index_key_t() = default;
+    index_key_t(const index_key_schema_t& key_schema, const uint8_t* payload);
 
     template <typename... T_keys>
     index_key_t(T_keys... keys);
diff --git a/production/db/inc/index/range_index.hpp b/production/db/inc/index/range_index.hpp
index b65411b921ea..15d6da02cbdd 100644
--- a/production/db/inc/index/range_index.hpp
+++ b/production/db/inc/index/range_index.hpp
@@ -28,8 +28,8 @@ using range_index_iterator_t = index_iterator_t<range_type_t, range_type_t::cons
 class range_index_t : public index_t<range_type_t, range_index_iterator_t>
 {
 public:
-    range_index_t(gaia::common::gaia_id_t index_id, common::gaia_type_t table_type, bool is_unique = false)
-        : index_t(index_id, catalog::index_type_t::range, table_type, is_unique)
+    range_index_t(gaia::common::gaia_id_t index_id, index_key_schema_t key_schema, bool is_unique = false)
+        : index_t(index_id, catalog::index_type_t::range, key_schema, is_unique)
     {
     }
     ~range_index_t() = default;
diff --git a/production/db/inc/query_processor/index_scan_physical.inc b/production/db/inc/query_processor/index_scan_physical.inc
index 62a7f9eff7a0..949a36c392b0 100644
--- a/production/db/inc/query_processor/index_scan_physical.inc
+++ b/production/db/inc/query_processor/index_scan_physical.inc
@@ -145,7 +145,7 @@ template <typename T_index, typename T_index_iterator>
 db::index::index_key_t index_scan_physical_t<T_index, T_index_iterator>::record_to_key(const db::index::index_record_t& record) const
 {
     auto obj = db::locator_to_ptr(record.locator);
-    return db::index::index_builder_t::make_key(m_index_id, obj->type, reinterpret_cast<const uint8_t*>(obj->data()));
+    return db::index::index_builder_t::make_key(m_index, reinterpret_cast<const uint8_t*>(obj->data()));
 }
 
 template <typename T_index, typename T_index_iterator>
diff --git a/production/db/index/CMakeLists.txt b/production/db/index/CMakeLists.txt
index 43a9c486e4d4..127a386d52a5 100644
--- a/production/db/index/CMakeLists.txt
+++ b/production/db/index/CMakeLists.txt
@@ -18,7 +18,8 @@ add_library(gaia_index STATIC
   src/base_index.cpp
   src/index.cpp
   src/hash_index.cpp
-  src/range_index.cpp)
+  src/range_index.cpp
+  src/index_offset_buffer.cpp)
 configure_gaia_target(gaia_index)
 target_include_directories(gaia_index PUBLIC ${GAIA_INDEX_INCLUDES})
 target_link_libraries(gaia_index PRIVATE gaia_common gaia_payload_types)
diff --git a/production/db/index/src/base_index.cpp b/production/db/index/src/base_index.cpp
index b954221e78a0..7436f19dd0b1 100644
--- a/production/db/index/src/base_index.cpp
+++ b/production/db/index/src/base_index.cpp
@@ -24,7 +24,12 @@ catalog::index_type_t base_index_t::type() const
 
 common::gaia_type_t base_index_t::table_type() const
 {
-    return m_table_type;
+    return m_key_schema.table_type;
+}
+
+const index_key_schema_t& base_index_t::key_schema() const
+{
+    return m_key_schema;
 }
 
 bool base_index_t::is_unique() const
diff --git a/production/db/index/src/index.cpp b/production/db/index/src/index.cpp
index 261a07e97719..b05238a1c3f9 100644
--- a/production/db/index/src/index.cpp
+++ b/production/db/index/src/index.cpp
@@ -22,6 +22,14 @@ namespace db
 namespace index
 {
 
+index_key_t::index_key_t(const index_key_schema_t& key_schema, const uint8_t* payload)
+{
+    for (field_position_t pos : key_schema.field_positions)
+    {
+        insert(payload_types::get_field_value(key_schema.table_type, payload, key_schema.binary_schema->data(), key_schema.binary_schema->size(), pos));
+    }
+}
+
 int index_key_t::compare(const index_key_t& other) const
 {
     ASSERT_PRECONDITION(
diff --git a/production/db/index/src/index_offset_buffer.cpp b/production/db/index/src/index_offset_buffer.cpp
new file mode 100644
index 000000000000..0beb2f4cb823
--- /dev/null
+++ b/production/db/index/src/index_offset_buffer.cpp
@@ -0,0 +1,42 @@
+/////////////////////////////////////////////
+// Copyright (c) Gaia Platform LLC
+// All rights reserved.
+/////////////////////////////////////////////
+
+#include "index.hpp"
+
+namespace gaia
+{
+namespace db
+{
+namespace index
+{
+
+gaia_offset_t index_offset_buffer_t::get_offset(size_t index) const
+{
+    return m_offsets[index].first;
+}
+
+common::gaia_type_t index_offset_buffer_t::get_type(size_t index) const
+{
+    return m_offsets[index].second;
+}
+
+size_t index_offset_buffer_t::size() const
+{
+    return m_size;
+}
+
+bool index_offset_buffer_t::empty() const
+{
+    return m_size == 0;
+}
+
+void index_offset_buffer_t::insert(gaia_offset_t offset, common::gaia_type_t type)
+{
+    m_offsets[m_size] = std::make_pair<gaia_offset_t, common::gaia_type_t>(std::move(offset), std::move(type));
+    ++m_size;
+}
+} // namespace index
+} // namespace db
+} // namespace gaia
diff --git a/production/db/index/tests/test_index.cpp b/production/db/index/tests/test_index.cpp
index ed12beafbde8..a0e2859ddec3 100644
--- a/production/db/index/tests/test_index.cpp
+++ b/production/db/index/tests/test_index.cpp
@@ -34,7 +34,8 @@ index_record_t create_index_record()
 
 TEST(index, empty_range_index)
 {
-    range_index_t range_index(0, 0);
+    index_key_schema_t schema;
+    range_index_t range_index(0, schema);
     size_t count = 0;
 
     for (auto it = range_index.begin(); it != range_index.end(); it++)
@@ -47,7 +48,8 @@ TEST(index, empty_range_index)
 
 TEST(index, empty_hash_index)
 {
-    hash_index_t hash_index(0, 0);
+    index_key_schema_t schema;
+    hash_index_t hash_index(0, schema);
     size_t count = 0;
     for (auto it = hash_index.begin(); it != hash_index.end(); it++)
     {
@@ -58,7 +60,8 @@ TEST(index, empty_hash_index)
 
 TEST(index, one_record_hash_index)
 {
-    hash_index_t hash_index(0, 0);
+    index_key_schema_t schema;
+    hash_index_t hash_index(0, schema);
     size_t count = 0;
 
     hash_index.insert_index_entry(index_key_t(0), index_record_t(create_index_record()));
@@ -74,7 +77,8 @@ TEST(index, one_record_hash_index)
 
 TEST(index, one_record_range_index)
 {
-    range_index_t range_index(0, 0);
+    index_key_schema_t schema;
+    range_index_t range_index(0, schema);
     size_t count = 0;
 
     range_index.insert_index_entry(index_key_t(0), index_record_t(create_index_record()));
@@ -90,7 +94,8 @@ TEST(index, one_record_range_index)
 
 TEST(index, single_key_multi_record_hash_index)
 {
-    hash_index_t hash_index(0, 0);
+    index_key_schema_t schema;
+    hash_index_t hash_index(0, schema);
     size_t count = 0;
 
     hash_index.insert_index_entry(index_key_t(0), index_record_t(create_index_record()));
@@ -108,7 +113,8 @@ TEST(index, single_key_multi_record_hash_index)
 
 TEST(index, single_key_multi_record_range_index)
 {
-    range_index_t range_index(0, 0);
+    index_key_schema_t schema;
+    range_index_t range_index(0, schema);
     size_t count = 0;
 
     range_index.insert_index_entry(index_key_t(0), index_record_t(create_index_record()));
@@ -126,7 +132,8 @@ TEST(index, single_key_multi_record_range_index)
 
 TEST(index, multi_key_multi_record_hash_index)
 {
-    hash_index_t hash_index(0, 0);
+    index_key_schema_t schema;
+    hash_index_t hash_index(0, schema);
     size_t count = 0;
 
     hash_index.insert_index_entry(index_key_t(0), index_record_t(create_index_record()));
@@ -146,7 +153,8 @@ TEST(index, multi_key_multi_record_hash_index)
 
 TEST(index, multi_key_multi_record_range_index)
 {
-    range_index_t range_index(0, 0);
+    index_key_schema_t schema;
+    range_index_t range_index(0, schema);
     size_t count = 0;
 
     range_index.insert_index_entry(index_key_t(0), index_record_t(create_index_record()));
@@ -167,8 +175,9 @@ TEST(index, multi_key_multi_record_range_index)
 // Simulate index updates for range index
 TEST(index, range_update_test)
 {
+    index_key_schema_t schema;
     index_record_t to_update = create_index_record();
-    range_index_t range_index(0, 0);
+    range_index_t range_index(0, schema);
     size_t count = 0;
 
     range_index.insert_index_entry(index_key_t(0), to_update);
@@ -190,7 +199,8 @@ TEST(index, range_update_test)
 TEST(index, hash_update_test)
 {
     index_record_t to_update = create_index_record();
-    hash_index_t hash_index(0, 0);
+    index_key_schema_t schema;
+    hash_index_t hash_index(0, schema);
     size_t count = 0;
 
     hash_index.insert_index_entry(index_key_t(0), to_update);
@@ -210,7 +220,8 @@ TEST(index, hash_update_test)
 
 TEST(index, eq_range_hash_index)
 {
-    hash_index_t hash_index(0, 0);
+    index_key_schema_t schema;
+    hash_index_t hash_index(0, schema);
     size_t count = 0;
 
     hash_index.insert_index_entry(index_key_t(0), index_record_t(create_index_record()));
@@ -232,7 +243,8 @@ TEST(index, eq_range_hash_index)
 
 TEST(index, eq_range_range_index)
 {
-    range_index_t range_index(0, 0);
+    index_key_schema_t schema;
+    range_index_t range_index(0, schema);
     size_t count = 0;
 
     range_index.insert_index_entry(index_key_t(0), index_record_t(create_index_record()));
diff --git a/production/db/memory_manager/CMakeLists.txt b/production/db/memory_manager/CMakeLists.txt
index 24d172100fb0..79d5b35691ae 100644
--- a/production/db/memory_manager/CMakeLists.txt
+++ b/production/db/memory_manager/CMakeLists.txt
@@ -8,7 +8,6 @@ project(memory_manager)
 set(GAIA_MEMORY_MANAGER_INCLUDES
   "${GAIA_INC}"
   "${GAIA_DB_INC}/memory_manager"
-  "${GAIA_DB_INC}/index"
   "${GAIA_REPO}/production/db/core/inc")
 
 message(VERBOSE "GAIA_MEMORY_MANAGER_INCLUDES=${GAIA_MEMORY_MANAGER_INCLUDES}")
diff --git a/production/inc/gaia_internal/db/index_builder.hpp b/production/inc/gaia_internal/db/index_builder.hpp
index f9584b51c36f..9b72650779ac 100644
--- a/production/inc/gaia_internal/db/index_builder.hpp
+++ b/production/inc/gaia_internal/db/index_builder.hpp
@@ -36,13 +36,13 @@ class index_builder_t
     static indexes_t::iterator create_empty_index(const index_view_t& index_view, bool skip_catalog_integrity_check = false);
     static void drop_index(const index_view_t& index_view);
     static void update_index(
-        common::gaia_id_t index_id, common::gaia_type_t type_id, const txn_log_t::log_record_t& log_record, bool allow_create_empty = false);
+        db_index_t index, const txn_log_t::log_record_t& log_record);
 
-    static index_key_t make_key(common::gaia_id_t index_id, common::gaia_type_t type_id, const uint8_t* payload);
+    static index_key_t make_key(db_index_t index, const uint8_t* payload);
     static void serialize_key(const index_key_t& key, payload_types::data_write_buffer_t& buffer);
     static index_key_t deserialize_key(common::gaia_id_t index_id, payload_types::data_read_buffer_t& buffer);
 
-    static void populate_index(common::gaia_id_t index_id, common::gaia_type_t type, gaia_locator_t locator);
+    static void populate_index(common::gaia_id_t index_id, gaia_locator_t locator);
     static void update_indexes_from_txn_log(const txn_log_t& records, bool skip_catalog_integrity_check, bool allow_create_empty = false);
     static void gc_indexes_from_txn_log(const txn_log_t& records, bool deallocate_new_offsets);
     static void mark_index_entries_committed(gaia_txn_id_t txn_id);
@@ -50,7 +50,7 @@ class index_builder_t
 private:
     static index_record_t make_record(gaia_locator_t locator, gaia_offset_t offset, index_record_operation_t operation);
 
-    static void update_index(common::gaia_id_t index_id, index_key_t&& key, index_record_t record, bool allow_create_empty = false);
+    static void update_index(db_index_t index, index_key_t&& key, index_record_t record);
 };
 
 } // namespace index