lbug 0.22.0

An in-process property graph database management system built for query speed and scalability
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138

#include "storage/wal/wal_record.h"

#include <algorithm>

#include "common/exception/runtime.h"
#include "common/serializer/buffer_writer.h"
#include "common/serializer/deserializer.h"
#include "common/serializer/serializer.h"
#include "main/client_context.h"

using namespace lbug::common;

namespace lbug {
namespace storage {

void WALRecord::serialize(Serializer& serializer) const {
    serializer.writeDebuggingInfo("type");
    serializer.write(type);
}

void WALRecord::serializeWithLength(Serializer& serializer, const WALRecord& record) {
    auto bufferWriter = std::make_shared<BufferWriter>();
    Serializer bufferSerializer{bufferWriter};
    record.serialize(bufferSerializer);
    bufferSerializer.writeDebuggingInfo("ownerCatalogName");
    bufferSerializer.write<std::string>(record.ownerCatalogName);

    const auto recordLength = bufferWriter->getSize();
    serializer.write(recordLength);
    serializer.write(bufferWriter->getBlobData(), recordLength);
}

std::unique_ptr<WALRecord> WALRecord::deserialize(Deserializer& deserializer,
    const main::ClientContext& clientContext) {
    std::string key;
    auto type = WALRecordType::INVALID_RECORD;
    deserializer.getReader()->onObjectBegin();
    uint64_t recordLength = 0;
    deserializer.deserializeValue(recordLength);
    deserializer.beginReadLimit(recordLength);
    deserializer.validateDebuggingInfo(key, "type");
    deserializer.deserializeValue(type);
    std::unique_ptr<WALRecord> walRecord;
    switch (type) {
    case WALRecordType::BEGIN_TRANSACTION_RECORD: {
        walRecord = BeginTransactionRecord::deserialize(deserializer);
    } break;
    case WALRecordType::COMMIT_RECORD: {
        walRecord = CommitRecord::deserialize(deserializer);
    } break;
    case WALRecordType::CREATE_CATALOG_ENTRY_RECORD: {
        walRecord = CreateCatalogEntryRecord::deserialize(deserializer);
    } break;
    case WALRecordType::CREATE_INDEX_RECORD: {
        walRecord = CreateIndexRecord::deserialize(deserializer);
    } break;
    case WALRecordType::DROP_CATALOG_ENTRY_RECORD: {
        walRecord = DropCatalogEntryRecord::deserialize(deserializer);
    } break;
    case WALRecordType::ALTER_TABLE_ENTRY_RECORD: {
        walRecord = AlterTableEntryRecord::deserialize(deserializer);
    } break;
    case WALRecordType::TABLE_INSERTION_RECORD: {
        walRecord = TableInsertionRecord::deserialize(deserializer, clientContext);
    } break;
    case WALRecordType::NODE_DELETION_RECORD: {
        walRecord = NodeDeletionRecord::deserialize(deserializer, clientContext);
    } break;
    case WALRecordType::NODE_UPDATE_RECORD: {
        walRecord = NodeUpdateRecord::deserialize(deserializer, clientContext);
    } break;
    case WALRecordType::REL_DELETION_RECORD: {
        walRecord = RelDeletionRecord::deserialize(deserializer, clientContext);
    } break;
    case WALRecordType::REL_DETACH_DELETE_RECORD: {
        walRecord = RelDetachDeleteRecord::deserialize(deserializer, clientContext);
    } break;
    case WALRecordType::REL_UPDATE_RECORD: {
        walRecord = RelUpdateRecord::deserialize(deserializer, clientContext);
    } break;
    case WALRecordType::COPY_TABLE_RECORD: {
        walRecord = CopyTableRecord::deserialize(deserializer);
    } break;
    case WALRecordType::CHECKPOINT_RECORD: {
        walRecord = CheckpointRecord::deserialize(deserializer);
    } break;
    case WALRecordType::UPDATE_SEQUENCE_RECORD: {
        walRecord = UpdateSequenceRecord::deserialize(deserializer);
    } break;
    case WALRecordType::UPDATE_SEQUENCE_NAMED_RECORD: {
        walRecord = UpdateSequenceNamedRecord::deserialize(deserializer);
    } break;
    case WALRecordType::LOAD_EXTENSION_RECORD: {
        walRecord = LoadExtensionRecord::deserialize(deserializer);
    } break;
    default: {
        throw RuntimeException("Corrupted wal file. Read out invalid WAL record type.");
    }
    }
    bool hasOwnerTrailer = deserializer.hasRemainingData();
    bool hasCompleteLengthField = false;
    uint64_t declaredOwnerNameLength = 0;
    uint64_t decodedOwnerNameLength = 0;
    bool hasTrailingInFrame = false;
    if (hasOwnerTrailer) {
        deserializer.validateDebuggingInfo(key, "ownerCatalogName");
        hasCompleteLengthField = deserializer.getRemainingReadLimit() >= sizeof(uint64_t);
        deserializer.deserializeValue(declaredOwnerNameLength);
        decodedOwnerNameLength =
            std::min(declaredOwnerNameLength, deserializer.getRemainingReadLimit());
        auto& ownerCatalogName = walRecord->ownerCatalogName;
        ownerCatalogName.resize(decodedOwnerNameLength);
        deserializer.read(reinterpret_cast<uint8_t*>(ownerCatalogName.data()),
            decodedOwnerNameLength);
        hasTrailingInFrame = deserializer.hasRemainingData();
    }
    walRecord->type = type;
    deserializer.skipReadLimit();
    deserializer.getReader()->onObjectEnd();
    if (hasOwnerTrailer) {
        if (!hasCompleteLengthField || declaredOwnerNameLength != decodedOwnerNameLength) {
            throw RuntimeException(
                "Corrupted wal file. Owner catalog name length overflows the record boundary.");
        }
        const auto& ownerCatalogName = walRecord->ownerCatalogName;
        if (ownerCatalogName.find('\0') != std::string::npos) {
            throw RuntimeException("Corrupted wal file. Owner catalog name contains a null byte.");
        }
        if (hasTrailingInFrame) {
            throw RuntimeException(
                "Corrupted wal file. Trailing bytes after the owner catalog name.");
        }
    }
    return walRecord;
}

} // namespace storage
} // namespace lbug