cache.hpp

/*
 * LMDB Abstraction Layer.
 * Copyright (C) 2023  Yury Gubich <blue@macaw.me>
 *
 * This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program.  If not, see <http://www.gnu.org/licenses/>.
 */
 
#pragma once
 
#include "cache.h"
#include "exceptions.h"


template<class K, class V>
LMDBAL::Cache<K, V>::Cache(Base* parent, const std::string& name, bool duplicates):
    Storage<K, V>(parent, name, duplicates),
    mode(Mode::nothing),
    cache(new std::map<K, V>()),
    abscent(new std::set<K>()),
    sizeDifference(0),
    transactionCache(new TransactionCache) {}

namespace LMDBAL {
template<class K, class V>
Cache<K, V>::~Cache() {
    delete transactionCache;
    delete cache;
    delete abscent;
}
}
 
template<class K, class V>
void LMDBAL::Cache<K, V>::addRecord(const K& key, const V& value) {
    StorageCommon::ensureOpened(StorageCommon::addRecordMethodName);
 
    if (cache->count(key) > 0)
        StorageCommon::throwDuplicate(StorageCommon::toString(key));
 
    Storage<K, V>::addRecord(key, value);
    handleAddRecord(key, value);
}
 
template<class K, class V>
void LMDBAL::Cache<K, V>::addRecord(const K& key, const V& value, TransactionID txn) {
    if (cache->count(key) > 0)
        StorageCommon::throwDuplicate(StorageCommon::toString(key));
 
    Storage<K, V>::addRecord(key, value, txn);
 
    typename TransactionCache::iterator tc = transactionCache->find(txn);
    if (tc != transactionCache->end()) {
        std::pair<K, V>* pair = new std::pair<K, V>(key, value);
        tc->second.emplace_back(Operation::add, pair);
    }
}
 
template<class K, class V>
void LMDBAL::Cache<K, V>::handleAddRecord(const K& key, const V& value) {
    cache->insert(std::make_pair(key, value));
 
    if (mode != Mode::full)
        abscent->erase(key);
}
 
template<class K, class V>
bool LMDBAL::Cache<K, V>::forceRecord(const K& key, const V& value) {
    StorageCommon::ensureOpened(StorageCommon::forceRecordMethodName);
 
    bool added = Storage<K, V>::forceRecord(key, value);
    handleForceRecord(key, value, added);
 
    return added;
}
 
template<class K, class V>
bool LMDBAL::Cache<K, V>::forceRecord(const K& key, const V& value, TransactionID txn) {
    bool added = Storage<K, V>::forceRecord(key, value, txn);
 
    typename TransactionCache::iterator tc = transactionCache->find(txn);
    if (tc != transactionCache->end()) {
        std::tuple<bool, K, V>* t = new std::tuple<bool, K, V>(added, key, value);
        tc->second.emplace_back(Operation::force, t);
    }
 
    return added;
}
 
template<class K, class V>
void LMDBAL::Cache<K, V>::handleForceRecord(const K& key, const V& value, bool added) {
    if (mode == Mode::full) {
        (*cache)[key] = value;
    } else {
        if (added)
            abscent->erase(key);
 
        std::pair<typename std::map<K, V>::iterator, bool> result =
            cache->insert(std::make_pair(key, value));
        if (!result.second)
            result.first->second = value;
        else if (!added)    //this way database had value but cache didn't, so, need to decrease sizeDifference
            handleMode();
    }
}
 
template<class K, class V>
void LMDBAL::Cache<K, V>::changeRecord(const K& key, const V& value) {
    StorageCommon::ensureOpened(StorageCommon::changeRecordMethodName);
 
    if (mode == Mode::full) {
        typename std::map<K, V>::iterator itr = cache->find(key);
        if (itr == cache->end())
            StorageCommon::throwNotFound(StorageCommon::toString(key));
 
        Storage<K, V>::changeRecord(key, value);
        itr->second = value;
    } else {
        if (abscent->count(key) > 0)
            StorageCommon::throwNotFound(StorageCommon::toString(key));
 
        try {
            Storage<K, V>::changeRecord(key, value);
            typename std::pair<typename std::map<K, V>::iterator, bool> res =
                cache->insert(std::make_pair(key, value));
            if (!res.second)
                res.first->second = value;
            else
                handleMode();
        } catch (const NotFound& error) {
            abscent->insert(key);
            throw error;
        }
    }
}
 
template<class K, class V>
void LMDBAL::Cache<K, V>::changeRecord(const K& key, const V& value, TransactionID txn) {
    if (mode == Mode::full) {
        typename std::map<K, V>::iterator itr = cache->find(key);
        if (itr == cache->end())
            StorageCommon::throwNotFound(StorageCommon::toString(key));
 
        Storage<K, V>::changeRecord(key, value, txn);
    } else {
        if (abscent->count(key) > 0)
            StorageCommon::throwNotFound(StorageCommon::toString(key));
 
        try {
            Storage<K, V>::changeRecord(key, value, txn);
        } catch (const NotFound& error) {
            abscent->insert(key);
            throw error;
        }
    }
 
    typename TransactionCache::iterator tc = transactionCache->find(txn);
    if (tc != transactionCache->end()) {
        std::pair<K, V>* pair = new std::pair<K, V>(key, value);
        tc->second.emplace_back(Operation::add, pair);
    }
}
 
template<class K, class V>
void LMDBAL::Cache<K, V>::handleChangeRecord(const K& key, const V& value) {
    if (mode == Mode::full) {
        cache->at(key) = value;
    } else {
        typename std::pair<typename std::map<K, V>::iterator, bool> res =
            cache->insert(std::make_pair(key, value));
        if (!res.second)
            res.first->second = value;
        else
            handleMode();
    }
}
 
template<class K, class V>
V LMDBAL::Cache<K, V>::getRecord(const K& key) const {
    StorageCommon::ensureOpened(StorageCommon::getRecordMethodName);
 
    V value;
    Cache<K, V>::getRecord(key, value);
    return value;
}
 
template<class K, class V>
void LMDBAL::Cache<K, V>::getRecord(const K& key, V& out) const {
    StorageCommon::ensureOpened(StorageCommon::getRecordMethodName);
 
    typename std::map<K, V>::const_iterator itr = cache->find(key);
    if (itr != cache->end()) {
        out = itr->second;
        return;
    }
 
    if (mode == Mode::full || abscent->count(key) != 0)
        StorageCommon::throwNotFound(StorageCommon::toString(key));
 
    try {
        Storage<K, V>::getRecord(key, out);
        appendToCache(key, out);
        return;
    } catch (const NotFound& error) {
        if (mode != Mode::full)
            abscent->insert(key);
 
        throw error;
    }
}
 
template<class K, class V>
V LMDBAL::Cache<K, V>::getRecord(const K& key, TransactionID txn) const {
    V value;
    Cache<K, V>::getRecord(key, value, txn);
    return value;
}
 
template<class K, class V>
void LMDBAL::Cache<K, V>::getRecord(const K& key, V& out, TransactionID txn) const {
    //if there are any changes made within this transaction
    //I will be able to see them among pending changes
    //so, I'm going to go through them in reverse order
    //and check every key. If it has anything to do this requested key
    //there is a way to tell...
    bool currentTransaction = false;
    typename TransactionCache::const_iterator tc = transactionCache->find(txn);
    if (tc != transactionCache->end()) {
        currentTransaction = true;
        const Queue& queue = tc->second;
        for (typename Queue::const_reverse_iterator i = queue.rbegin(), end = queue.rend(); i != end; ++i) {
            const Entry& entry = *i;
 
            switch (entry.first) {
                case Operation::add:
                    if (static_cast<std::pair<K, V>*>(entry.second)->first == key) {
                        out = static_cast<std::pair<K, V>*>(entry.second)->second;
                        return;
                    }
                    break;
                case Operation::remove:
                        StorageCommon::throwNotFound(StorageCommon::toString(key));
                    break;
                case Operation::change:
                    if (static_cast<std::pair<K, V>*>(entry.second)->first == key) {
                        out = static_cast<std::pair<K, V>*>(entry.second)->second;
                        return;
                    }
 
                    break;
                case Operation::force:
                    if (std::get<1>(*static_cast<std::tuple<bool, K, V>*>(entry.second)) == key) {
                        out = std::get<2>(*static_cast<std::tuple<bool, K, V>*>(entry.second));
                        return;
                    }
                    break;
                case Operation::drop:
                    StorageCommon::throwNotFound(StorageCommon::toString(key));
                    break;
                case Operation::replace: {
                    std::map<K, V>* newMap = static_cast<std::map<K, V>*>(entry.second);
                    typename std::map<K, V>::const_iterator vitr = newMap->find(key);
                    if (vitr != newMap->end()) {
                        out = vitr->second;
                        return;
                    } else {
                        StorageCommon::throwNotFound(StorageCommon::toString(key));
                    }
                }
                    break;
                case Operation::addMany: {
                    const std::tuple<bool, SizeType, std::map<K, V>>& tuple =
                        *static_cast<std::tuple<bool, SizeType, std::map<K, V>>*>(entry.second);
                    const std::map<K, V>& newElements = std::get<2>(tuple);
                    typename std::map<K, V>::const_iterator vitr = newElements.find(key);
                    if (vitr != newElements.end()) {
                        out = vitr->second;
                        return;
                    }
                }
                    break;
            }
        }
    }
    //... but if nothing was found or if the transaction is not the one
    //which caused the changes i just need to check it among local cache
 
    typename std::map<K, V>::const_iterator itr = cache->find(key);
    if (itr != cache->end()) {
        out = itr->second;
        return;
    }
 
    if (mode == Mode::full || abscent->count(key) != 0)
        StorageCommon::throwNotFound(StorageCommon::toString(key));
 
    try {
        Storage<K, V>::getRecord(key, out, txn);
        if (!currentTransaction)
            appendToCache(key, out);
 
        return;
    } catch (const NotFound& error) {
        if (!currentTransaction && mode != Mode::full)
            abscent->insert(key);
 
        throw error;
    }
}
 
template<class K, class V>
void LMDBAL::Cache<K, V>::discoveredRecord(const K& key, const V& value) const {
    appendToCache(key, value);
}
 
template<class K, class V>
void LMDBAL::Cache<K, V>::discoveredRecord(const K& key, const V& value, TransactionID txn) const {
    typename TransactionCache::const_iterator tc = transactionCache->find(txn);
    if (tc == transactionCache->end())      //there is a way to look though all the records in transaction log and cache the new pair
        discoveredRecord(key, value);       //if there is nothing in transaction log about it, but it seems like too much for a small gain
}
 
template<class K, class V>
bool LMDBAL::Cache<K, V>::checkRecord(const K& key) const {
    StorageCommon::ensureOpened(StorageCommon::checkRecordMethodName);
 
    typename std::map<K, V>::const_iterator itr = cache->find(key);
    if (itr != cache->end())
        return true;
 
    if (mode == Mode::full || abscent->count(key) != 0)
        return false;
 
    try {
        V value = Storage<K, V>::getRecord(key);
        appendToCache(key, value);
 
        return true;
    } catch (const NotFound& error) {
        if (mode != Mode::full)
            abscent->insert(key);
 
        return false;
    }
}
 
template<class K, class V>
bool LMDBAL::Cache<K, V>::checkRecord(const K& key, TransactionID txn) const {
    //if there are any changes made within this transaction
    //I will be able to see them among pending changes
    //so, I'm going to go through them in reverse order
    //and check every key. If it has anything to do this requested key
    //there is a way to tell...
    bool currentTransaction = false;
    typename TransactionCache::const_iterator tc = transactionCache->find(txn);
    if (tc != transactionCache->end()) {
        currentTransaction = true;
        const Queue& queue = tc->second;
        for (typename Queue::const_reverse_iterator i = queue.rbegin(), end = queue.rend(); i != end; ++i) {
            const Entry& entry = *i;
 
            switch (entry.first) {
                case Operation::add:
                    if (static_cast<std::pair<K, V>*>(entry.second)->first == key)
                        return true;
                    break;
                case Operation::remove:
                    if (*static_cast<K*>(entry.second) == key)
                        return false;
                    break;
                case Operation::change:
                    if (static_cast<std::pair<K, V>*>(entry.second)->first == key)
                        return true;
                    break;
                case Operation::force:
                    if (std::get<1>(*static_cast<std::tuple<bool, K, V>*>(entry.second)) == key)
                        return true;
                    break;
                case Operation::drop:
                    return false;
                    break;
                case Operation::replace:
                    if (static_cast<std::map<K, V>*>(entry.second)->count(key) > 0)
                        return true;
                    else
                        return false;
                    break;
                case Operation::addMany:
                    if (std::get<2>(
                            *static_cast<std::tuple<bool, SizeType, std::map<K, V>>*>(entry.second)
                        ).count(key) > 0)
                        return true;
                    break;
            }
        }
    }
    //... but if nothing was found or if the transaction is not the one
    //which caused the changes i just need to check it among local cache
 
    typename std::map<K, V>::const_iterator itr = cache->find(key);
    if (itr != cache->end())
        return true;
 
    if (mode == Mode::full || abscent->count(key) != 0)
        return false;
 
    try {
        V value = Storage<K, V>::getRecord(key, txn);
        if (!currentTransaction)
            appendToCache(key, value);
 
        return true;
    } catch (const NotFound& error) {
        if (!currentTransaction && mode != Mode::full)
            abscent->insert(key);
 
        return false;
    }
}
 
template<class K, class V>
void LMDBAL::Cache<K, V>::appendToCache(const K& key, const V& value) const {
    typename std::pair<typename std::map<K, V>::const_iterator, bool> pair = cache->insert(std::make_pair(key, value));
    if (pair.second)
        handleMode();
}
 
template<class K, class V>
std::map<K, V> LMDBAL::Cache<K, V>::readAll() const {
    StorageCommon::ensureOpened(StorageCommon::readAllMethodName);
 
    if (mode != Mode::full) {               //there is a room for optimization
        mode = Mode::full;                  //I can read and deserialize only those values
        *cache = Storage<K, V>::readAll();  //that are missing in the cache
        abscent->clear();
        sizeDifference = 0;
    }
 
    return *cache;
}
 
template<class K, class V>
void LMDBAL::Cache<K, V>::readAll(std::map<K, V>& out) const {
    StorageCommon::ensureOpened(StorageCommon::readAllMethodName);
 
    if (mode != Mode::full) {               //there is a room for optimization
        mode = Mode::full;                  //I can read and deserialize only those values
        Storage<K, V>::readAll(out);        //that are missing in the cache
        *cache = out;
        abscent->clear();
        sizeDifference = 0;
    }
}
 
template<class K, class V>
std::map<K, V> LMDBAL::Cache<K, V>::readAll(TransactionID txn) const {
    std::map<K, V> out;
    readAll(out, txn);
 
    return out;
}
 
template<class K, class V>
void LMDBAL::Cache<K, V>::readAll(std::map<K, V>& out, TransactionID txn) const {
    typename TransactionCache::iterator tc = transactionCache->find(txn);
    if (tc != transactionCache->end()) {
        Queue& queue = tc->second;
        if (mode != Mode::full) {
            out = *cache;
 
            for (typename Queue::const_iterator i = queue.begin(), end = queue.end(); i != end; ++i) {
                const Entry& entry = *i;
                switch (entry.first) {
                    case Operation::add:
                        out.insert(*static_cast<std::pair<K, V>*>(entry.second));
                        break;
                    case Operation::remove:
                        out.erase(*static_cast<K*>(entry.second));
                        break;
                    case Operation::change: {
                        std::pair<K, V>* pair = static_cast<std::pair<K, V>*>(entry.second);
                        out.at(pair->first) = pair->second;
                    }
                        break;
                    case Operation::force:{
                        const std::tuple<bool, K, V>& tuple =
                            *static_cast<std::tuple<bool, K, V>*>(entry.second);
                        out[std::get<1>(tuple)] = std::get<2>(tuple);
                    }
                        break;
                    case Operation::drop:
                        out.clear();
                        break;
                    case Operation::replace:
                        out = *static_cast<std::map<K, V>*>(entry.second);
                        break;
                    case Operation::addMany: {
                        const std::tuple<bool, SizeType, std::map<K, V>>& t =
                            *static_cast<std::tuple<bool, SizeType, std::map<K, V>>*>(entry.second);
                        const std::map<K, V>& added = std::get<2>(t);
                        bool overwrite = std::get<0>(t);
                        for (const std::pair<const K, V>& pair : added) {
                            if (overwrite)
                                out[pair.first] = pair.second;
                            else
                                out.insert(pair);
                        }
                    }
                        break;
                }
            }
        } else {
            Storage<K, V>::readAll(out, txn);
 
            //queue.clear();                                                    //since I'm getting a complete state of the database
            queue.emplace_back(Operation::replace, new std::map<K, V>(out));    //I can as well erase all previous cache entries
        }
    } else {
        if (mode != Mode::full) {               //there is a room for optimization
            mode = Mode::full;                  //I can read and deserialize only those values
            Storage<K, V>::readAll(out, txn);   //that are missing in the cache
            *cache = out;
            abscent->clear();
            sizeDifference = 0;
        }
    }
}
 
template<class K, class V>
void LMDBAL::Cache<K, V>::replaceAll(const std::map<K, V>& data) {
    Storage<K, V>::replaceAll(data);
    *cache = data;
 
    if (mode != Mode::full) {
        mode = Mode::full;
        abscent->clear();
        sizeDifference = 0;
    }
}
 
template<class K, class V>
void LMDBAL::Cache<K, V>::replaceAll(const std::map<K, V>& data, TransactionID txn) {
    Storage<K, V>::replaceAll(data, txn);
 
    typename TransactionCache::iterator tc = transactionCache->find(txn);
    if (tc != transactionCache->end()) {
        //queue.clear();
        std::map<K, V>* map = new std::map<K, V>(data);     //since I'm getting a complete state of the database
        tc->second.emplace_back(Operation::replace, map);   //I can as well erase all previous cache entries
    }
}
 
template<class K, class V>
void LMDBAL::Cache<K, V>::handleReplaceAll(std::map<K, V>* data) {
    delete cache;
    cache = data;
 
    if (mode != Mode::full) {
        mode = Mode::full;
        abscent->clear();
        sizeDifference = 0;
    }
}
 
template<class K, class V>
LMDBAL::SizeType LMDBAL::Cache<K, V>::addRecords(const std::map<K, V>& data, bool overwrite) {
    SizeType newSize = Storage<K, V>::addRecords(data, overwrite);
    handleAddRecords(data, overwrite, newSize);
 
    return newSize;
}
 
template<class K, class V>
LMDBAL::SizeType LMDBAL::Cache<K, V>::addRecords(const std::map<K, V>& data, TransactionID txn, bool overwrite) {
    SizeType newSize = Storage<K, V>::addRecords(data, txn, overwrite);
 
    typename TransactionCache::iterator tc = transactionCache->find(txn);
    if (tc != transactionCache->end()) {
        std::tuple<bool, SizeType, std::map<K, V>>* tuple =
            new std::tuple<bool, SizeType, std::map<K, V>>(overwrite, newSize, data);
        tc->second.emplace_back(Operation::addMany, tuple);
    }
 
    return newSize;
}
 
template<class K, class V>
void LMDBAL::Cache<K, V>::handleAddRecords(const std::map<K, V>& data, bool overwrite, SizeType newSize) {
    if (mode == Mode::nothing)
        mode = Mode::size;
 
    std::map<K, V>& c = *cache;
    std::set<K>& a = *abscent;
    for (const std::pair<const K, V>& pair : data) {
        std::pair<typename std::map<K, V>::iterator, bool> res = c.insert(pair);
        if (!res.second) {
            if (overwrite)
                res.first->second = pair.second;
        } else if (mode != Mode::full) {
            a.erase(pair.first);
        }
    }
 
    if (mode != Mode::full) {
        sizeDifference = newSize - static_cast<SizeType>(c.size());
        if (sizeDifference == 0) {
            mode = Mode::full;
            abscent->clear();
        }
    }
}
 
template<class K, class V>
void LMDBAL::Cache<K, V>::removeRecord(const K& key) {
    StorageCommon::ensureOpened(StorageCommon::removeRecordMethodName);
 
    bool noKey = false;
    if (mode != Mode::full)
        noKey = cache->count(key) == 0;
    else
        noKey = abscent->count(key) > 0;
 
    if (noKey)
        StorageCommon::throwNotFound(StorageCommon::toString(key));
 
    Storage<K, V>::removeRecord(key);
    handleRemoveRecord(key);
}
 
template<class K, class V>
void LMDBAL::Cache<K, V>::removeRecord(const K& key, TransactionID txn) {
    bool noKey = false;
    if (mode != Mode::full)
        noKey = cache->count(key) == 0;
    else
        noKey = abscent->count(key) > 0;
 
    if (noKey)
        StorageCommon::throwNotFound(StorageCommon::toString(key));
 
    Storage<K, V>::removeRecord(key, txn);
 
    typename TransactionCache::iterator tc = transactionCache->find(txn);
    if (tc != transactionCache->end())
        tc->second.emplace_back(Operation::remove, new K(key));
}
 
template<class K, class V>
void LMDBAL::Cache<K, V>::handleRemoveRecord(const K& key)  {
    if (cache->erase(key) == 0)     //if it was not cached and we are now in size mode then the sizeDifference would decrease
        handleMode();
 
    if (mode != Mode::full)
        abscent->insert(key);
}
 
template<class K, class V>
LMDBAL::SizeType LMDBAL::Cache<K, V>::count() const {
    size_t cacheSize = cache->size();
    if (cacheSize > std::numeric_limits<LMDBAL::SizeType>::max())
        throw std::overflow_error("Cache size is too big");
 
    switch (mode) {
        case Mode::nothing:
        {
            SizeType sz = Storage<K, V>::count();
            sizeDifference = sz - static_cast<SizeType>(cacheSize);
            if (sz == 0) {
                mode = Mode::full;
                abscent->clear();
            } else {
                mode = Mode::size;
            }
            return sz;
        }
        case Mode::size:
            return static_cast<SizeType>(cacheSize) + sizeDifference;
        case Mode::full:
            return static_cast<SizeType>(cacheSize);
        default:
            return 0;   //unreachable, no such state, just to suppress the warning
    }
}
 
template<class K, class V>
LMDBAL::SizeType LMDBAL::Cache<K, V>::count(TransactionID txn) const {
    int32_t diff = 0;
    bool currentTransaction = false;
    typename TransactionCache::const_iterator tc = transactionCache->find(txn);
    if (tc != transactionCache->end()) {
        currentTransaction = true;
        const Queue& queue = tc->second;
        for (typename Queue::const_reverse_iterator i = queue.rbegin(), end = queue.rend(); i != end; ++i) {
            const Entry& entry = *i;
 
            switch (entry.first) {
                case Operation::add:
                    ++diff;
                    break;
                case Operation::remove:
                    --diff;
                    break;
                case Operation::change:
                    break;
                case Operation::force:
                    if (std::get<0>(*static_cast<std::tuple<bool, K, V>*>(entry.second)))
                        ++diff;
                    break;
                case Operation::drop:
                    return false;
                    break;
                case Operation::replace: {
                    auto result = static_cast<int32_t>(static_cast<std::map<K, V>*>(entry.second)->size()) + diff;
                    if (result < 0)
                        throw std::overflow_error("Invalid cache size");
 
                    return static_cast<SizeType>(result);
                }
                case Operation::addMany:{
                    auto result = static_cast<int32_t>(std::get<1>(*static_cast<std::tuple<bool, SizeType, std::map<K, V>>*>(entry.second))) + diff;
                    if (result < 0)
                        throw std::overflow_error("Invalid cache size");
 
                    return static_cast<SizeType>(result);
                }
            }
        }
    }
 
    size_t cacheSize = cache->size();
    if (cacheSize > std::numeric_limits<SizeType>::max())
        throw std::overflow_error("Cache size is too big");
 
    switch (mode) {
        case Mode::nothing: {
            SizeType sz = Storage<K, V>::count(txn);
            if (!currentTransaction) {
                sizeDifference = sz - static_cast<SizeType>(cacheSize);
                if (sz == 0) {
                    mode = Mode::full;
                    abscent->clear();
                } else {
                    mode = Mode::size;
                }
            }
            return sz;
        }
        case Mode::size: {
            auto result = static_cast<int32_t>(cacheSize) + static_cast<int32_t>(sizeDifference) + diff;
 
            if (result < 0)
                throw std::overflow_error("Invalid cache size");
            return static_cast<SizeType>(result);
        }
        case Mode::full: {
            auto result = static_cast<int32_t>(cacheSize) + diff;
 
            if (result < 0)
                throw std::overflow_error("Invalid cache size");
 
            return static_cast<SizeType>(result);
        }
        default:
            return 0;   //unreachable, no such state, just to suppress the waring
    }
}
 
template<class K, class V>
void LMDBAL::Cache<K, V>::handleMode() const {
    if (mode == Mode::size) {
        --sizeDifference;
        if (sizeDifference == 0) {
            mode = Mode::full;
            abscent->clear();
        }
    }
}
 
template<class K, class V>
int LMDBAL::Cache<K, V>::drop(const WriteTransaction& transaction) {
    StorageCommon::ensureOpened(StorageCommon::dropMethodName);
    TransactionID txn = StorageCommon::extractTransactionId(transaction, StorageCommon::dropMethodName);
    int res = Storage<K, V>::drop(txn);
 
    if (res != MDB_SUCCESS)
        return res;
 
    typename TransactionCache::iterator tc = transactionCache->find(txn);
    if (tc != transactionCache->end())
        tc->second.emplace_back(Operation::drop, nullptr);
 
    return res;
}
 
template<class K, class V>
void LMDBAL::Cache<K, V>::handleDrop() {
    cache->clear();
    abscent->clear();
    mode = Mode::full;
    sizeDifference = 0;
}
 
template<class K, class V>
void LMDBAL::Cache<K, V>::transactionStarted(TransactionID txn, bool readOnly) const {
    if (!readOnly)
        transactionCache->emplace(txn, Queue());
}
 
template<class K, class V>
void LMDBAL::Cache<K, V>::transactionCommited(TransactionID txn) {
    typename TransactionCache::iterator itr = transactionCache->find(txn);
    if (itr != transactionCache->end()) {
        Queue& queue = itr->second;
        for (const Entry& entry : queue)
            handleTransactionEntry(entry);
 
        transactionCache->erase(itr);
    }
}
 
template<class K, class V>
void LMDBAL::Cache<K, V>::transactionAborted(TransactionID txn) const {
    typename TransactionCache::iterator itr = transactionCache->find(txn);
    if (itr != transactionCache->end()) {
        Queue& queue = itr->second;
        for (const Entry& entry : queue)
            destroyTransactionEntry(entry);
 
        transactionCache->erase(itr);
    }
}
 
template<class K, class V>
void LMDBAL::Cache<K, V>::handleTransactionEntry(const Entry& entry) {
    switch (entry.first) {
        case Operation::add: {
            std::pair<K, V>* pair = static_cast<std::pair<K, V>*>(entry.second);
            handleAddRecord(pair->first, pair->second);
            delete pair;
        }   break;
        case Operation::remove: {
            K* key = static_cast<K*>(entry.second);
            handleRemoveRecord(*key);
            delete key;
        }   break;
        case Operation::change: {
            std::pair<K, V>* pair = static_cast<std::pair<K, V>*>(entry.second);
            handleChangeRecord(pair->first, pair->second);
            delete pair;
        }   break;
        case Operation::force: {
            std::tuple<bool, K, V>* tuple = static_cast<std::tuple<bool, K, V>*>(entry.second);
            const std::tuple<bool, K, V>& t = *tuple;
            handleForceRecord(std::get<1>(t), std::get<2>(t), std::get<0>(t));
            delete tuple;
        }   break;
        case Operation::drop:
            handleDrop();
            break;
        case Operation::replace:
            handleReplaceAll(static_cast<std::map<K, V>*>(entry.second));   //I take ownership, no need to delete
            break;
        case Operation::addMany: {
            std::tuple<bool, SizeType, std::map<K, V>>* tuple = static_cast<std::tuple<bool, SizeType, std::map<K, V>>*>(entry.second);
            const std::tuple<bool, SizeType, std::map<K, V>>& t = * tuple;
            handleAddRecords(std::get<2>(t), std::get<0>(t), std::get<1>(t));
            delete tuple;
        }   break;
    }
}
 
template<class K, class V>
void LMDBAL::Cache<K, V>::destroyTransactionEntry(const Entry& entry) const {
    switch (entry.first) {
        case Operation::add:
            delete static_cast<std::pair<K, V>*>(entry.second);
            break;
        case Operation::remove:
            delete static_cast<K*>(entry.second);
            break;
        case Operation::change:
            delete static_cast<std::pair<K, V>*>(entry.second);
            break;
        case Operation::force:
            delete static_cast<std::tuple<bool, K, V>*>(entry.second);
            break;
        case Operation::drop:
            break;
        case Operation::replace:
            delete static_cast<std::map<K, V>*>(entry.second);
            break;
        case Operation::addMany:
            delete static_cast<std::tuple<bool, SizeType, std::map<K, V>>*>(entry.second);
            break;
    }
}