undo_buffer.h

Go to the documentation of this file.

#pragma once
 
#include <mutex>
 
#include "constants.h"
#include "types.h"
 
namespace kuzu {
namespace catalog {
class CatalogEntry;
class CatalogSet;
class SequenceCatalogEntry;
struct SequenceRollbackData;
} // namespace catalog
namespace transaction {
class Transaction;
}
 
namespace main {
class ClientContext;
}
namespace storage {
 
// TODO(Guodong): This should be reworked to use MemoryManager for memory allocaiton.
// For now, we use malloc to get around the limitation of 256KB from MM.
class UndoMemoryBuffer {
public:
 static constexpr uint64_t UNDO_MEMORY_BUFFER_SIZE = common::BufferPoolConstants::PAGE_4KB_SIZE;
 
 explicit UndoMemoryBuffer(uint64_t size) : size{size} {
 data = std::make_unique<uint8_t[]>(size);
 currentPosition = 0;
 }
 
 uint8_t* getDataUnsafe() const { return data.get(); }
 uint8_t const* getData() const { return data.get(); }
 uint64_t getSize() const { return size; }
 uint64_t getCurrentPosition() const { return currentPosition; }
 void moveCurrentPosition(uint64_t offset) {
 KU_ASSERT(currentPosition + offset <= size);
 currentPosition += offset;
 }
 bool canFit(uint64_t size_) const { return currentPosition + size_ <= this->size; }
 
private:
 std::unique_ptr<uint8_t[]> data;
 uint64_t size;
 uint64_t currentPosition;
};
 
class UndoBuffer;
class UndoBufferIterator {
public:
 explicit UndoBufferIterator(const UndoBuffer& undoBuffer) : undoBuffer{undoBuffer} {}
 
 template<typename F>
 void iterate(F&& callback);
 template<typename F>
 void reverseIterate(F&& callback);
 
private:
 const UndoBuffer& undoBuffer;
};
 
class UpdateInfo;
class VersionInfo;
struct VectorUpdateInfo;
struct VectorVersionInfo;
class WAL;
// This class is not thread safe, as it is supposed to be accessed by a single thread.
class UndoBuffer {
 friend class UndoBufferIterator;
 
public:
 enum class UndoRecordType : uint16_t {
 CATALOG_ENTRY = 0,
 SEQUENCE_ENTRY = 1,
 UPDATE_INFO = 6,
 INSERT_INFO = 7,
 DELETE_INFO = 8,
 };
 
 explicit UndoBuffer(transaction::Transaction* transaction);
 
 void createCatalogEntry(catalog::CatalogSet& catalogSet, catalog::CatalogEntry& catalogEntry);
 void createSequenceChange(catalog::SequenceCatalogEntry& sequenceEntry,
 const catalog::SequenceRollbackData& data);
 void createVectorInsertInfo(VersionInfo* versionInfo, common::idx_t vectorIdx,
 common::row_idx_t startRowInVector, common::row_idx_t numRows);
 void createVectorDeleteInfo(VersionInfo* versionInfo, common::idx_t vectorIdx,
 common::row_idx_t startRowInVector, common::row_idx_t numRows);
 void createVectorUpdateInfo(UpdateInfo* updateInfo, common::idx_t vectorIdx,
 VectorUpdateInfo* vectorUpdateInfo);
 
 void commit(common::transaction_t commitTS) const;
 void rollback();
 
 uint64_t getMemUsage() const;
 
private:
 uint8_t* createUndoRecord(uint64_t size);
 
 void createVectorVersionInfo(UndoRecordType recordType, VersionInfo* versionInfo,
 common::idx_t vectorIdx, common::row_idx_t startRowInVector, common::row_idx_t numRows);
 
 void commitRecord(UndoRecordType recordType, const uint8_t* record,
 common::transaction_t commitTS) const;
 void rollbackRecord(UndoRecordType recordType, const uint8_t* record);
 
 void commitCatalogEntryRecord(const uint8_t* record, common::transaction_t commitTS) const;
 void rollbackCatalogEntryRecord(const uint8_t* record);
 
 void commitSequenceEntry(uint8_t const* entry, common::transaction_t commitTS) const;
 void rollbackSequenceEntry(uint8_t const* entry);
 
 void commitVectorVersionInfo(UndoRecordType recordType, const uint8_t* record,
 common::transaction_t commitTS) const;
 void rollbackVectorVersionInfo(UndoRecordType recordType, const uint8_t* record);
 
 void commitVectorUpdateInfo(const uint8_t* record, common::transaction_t commitTS) const;
 void rollbackVectorUpdateInfo(const uint8_t* record) const;
 
private:
 std::mutex mtx;
 transaction::Transaction* transaction;
 std::vector<UndoMemoryBuffer> memoryBuffers;
};
 
} // namespace storage
} // namespace kuzu