Separate shadow pages from wal records and rework wal to use serializer

CMakeLists.txt

@@ -1,6 +1,6 @@
 cmake_minimum_required(VERSION 3.15)
 
-project(Kuzu VERSION 0.3.2.5 LANGUAGES CXX C)
+project(Kuzu VERSION 0.3.2.6 LANGUAGES CXX C)
 
 find_package(Threads REQUIRED)
 

src/common/serializer/buffered_file.cpp

@@ -2,6 +2,8 @@
 
 #include <cstring>
 
+#include "common/assert.h"
+#include "common/exception/runtime.h"
 #include "common/file_system/file_info.h"
 
 namespace kuzu {
@@ -15,7 +17,9 @@ BufferedFileWriter::BufferedFileWriter(std::unique_ptr<FileInfo> fileInfo)
     : buffer(std::make_unique<uint8_t[]>(BUFFER_SIZE)), fileOffset(0), bufferOffset(0),
       fileInfo(std::move(fileInfo)) {}
 
+// TODO: write assumes size is always less than BUFFER_SIZE here. should change this when needed.
 void BufferedFileWriter::write(const uint8_t* data, uint64_t size) {
+    KU_ASSERT(size <= BUFFER_SIZE);
     if (bufferOffset + size <= BUFFER_SIZE) {
         memcpy(&buffer[bufferOffset], data, size);
         bufferOffset += size;
@@ -39,7 +43,8 @@ void BufferedFileWriter::flush() {
 
 BufferedFileReader::BufferedFileReader(std::unique_ptr<FileInfo> fileInfo)
     : buffer(std::make_unique<uint8_t[]>(BUFFER_SIZE)), fileOffset(0), bufferOffset(0),
-      fileInfo(std::move(fileInfo)) {
+      fileInfo(std::move(fileInfo)), bufferSize{0} {
+    fileSize = this->fileInfo->getFileSize();
     readNextPage();
 }
 
@@ -58,9 +63,17 @@ void BufferedFileReader::read(uint8_t* data, uint64_t size) {
     }
 }
 
+bool BufferedFileReader::finished() {
+    return bufferOffset >= bufferSize && fileSize <= fileOffset;
+}
+
 void BufferedFileReader::readNextPage() {
-    fileInfo->readFromFile(buffer.get(), BUFFER_SIZE, fileOffset);
-    fileOffset += BUFFER_SIZE;
+    bufferSize = std::min(fileSize - fileOffset, BUFFER_SIZE);
+    if (bufferSize == 0) {
+        throw RuntimeException(stringFormat("Reading past the end of the file {}", fileInfo->path));
+    }
+    fileInfo->readFromFile(buffer.get(), bufferSize, fileOffset);
+    fileOffset += bufferSize;
     bufferOffset = 0;
 }
 

src/include/common/cast.h

@@ -1,5 +1,7 @@
 #pragma once
 
+#include <typeinfo>
+
 #include "common/assert.h"
 
 namespace kuzu {
@@ -8,11 +10,15 @@ namespace common {
 template<typename FROM, typename TO>
 TO ku_dynamic_cast(FROM old) {
 #if defined(KUZU_RUNTIME_CHECKS) || !defined(NDEBUG)
-    TO newVal = dynamic_cast<TO>(old);
-    if constexpr (std::is_pointer<FROM>()) {
-        KU_ASSERT(newVal != nullptr);
+    try {
+        TO newVal = dynamic_cast<TO>(old);
+        if constexpr (std::is_pointer<FROM>()) {
+            KU_ASSERT(newVal != nullptr);
+        }
+        return newVal;
+    } catch (std::bad_cast& e) {
+        KU_ASSERT(false);
     }
-    return newVal;
 #else
     return reinterpret_cast<TO>(old);
 #endif

src/include/common/constants.h

@@ -84,6 +84,7 @@ struct BufferPoolConstants {
 struct StorageConstants {
     static constexpr char OVERFLOW_FILE_SUFFIX[] = ".ovf";
     static constexpr char WAL_FILE_SUFFIX[] = ".wal";
+    static constexpr char SHADOWING_SUFFIX[] = ".shadow";
     static constexpr char INDEX_FILE_SUFFIX[] = ".hindex";
     static constexpr char NODES_STATISTICS_AND_DELETED_IDS_FILE_NAME[] =
         "nodes.statistics_and_deleted.ids";

src/include/common/file_system/file_info.h

@@ -16,6 +16,7 @@ struct KUZU_API FileInfo {
 
     virtual ~FileInfo() = default;
 
+    // TODO: This function should be marked as const.
     uint64_t getFileSize();
 
     void readFromFile(void* buffer, uint64_t numBytes, uint64_t position);

src/include/common/serializer/buffered_file.h

@@ -1,6 +1,5 @@
 #pragma once
 
-#include <cstdint>
 #include <memory>
 
 #include "common/constants.h"
@@ -19,13 +18,22 @@ class BufferedFileWriter final : public Writer {
 
     void write(const uint8_t* data, uint64_t size) override;
 
+    void flush();
+
+    void resetOffsets() {
+        fileOffset = 0;
+        bufferOffset = 0;
+    }
+
+    uint64_t getFileOffset() const { return fileOffset; }
+
+    FileInfo& getFileInfo() { return *fileInfo; }
+
 protected:
     std::unique_ptr<uint8_t[]> buffer;
     uint64_t fileOffset, bufferOffset;
     std::unique_ptr<FileInfo> fileInfo;
     static const uint64_t BUFFER_SIZE = BufferPoolConstants::PAGE_4KB_SIZE;
-
-    void flush();
 };
 
 class BufferedFileReader final : public Reader {
@@ -34,12 +42,18 @@ class BufferedFileReader final : public Reader {
 
     void read(uint8_t* data, uint64_t size) override;
 
+    bool finished() override;
+
 private:
+    static constexpr uint64_t BUFFER_SIZE = BufferPoolConstants::PAGE_4KB_SIZE;
+
     std::unique_ptr<uint8_t[]> buffer;
     uint64_t fileOffset, bufferOffset;
     std::unique_ptr<FileInfo> fileInfo;
-    static const uint64_t BUFFER_SIZE = BufferPoolConstants::PAGE_4KB_SIZE;
+    uint64_t fileSize;
+    uint64_t bufferSize;
 
+private:
     void readNextPage();
 };
 

src/include/common/serializer/deserializer.h

@@ -1,6 +1,5 @@
 #pragma once
 
-#include <cstdint>
 #include <memory>
 #include <string>
 #include <unordered_map>
@@ -16,6 +15,8 @@ class Deserializer {
 public:
     explicit Deserializer(std::unique_ptr<Reader> reader) : reader(std::move(reader)) {}
 
+    bool finished() { return reader->finished(); }
+
     template<typename T>
         requires std::is_trivially_destructible<T>::value || std::is_same<std::string, T>::value
     void deserializeValue(T& value) {

src/include/common/serializer/reader.h

@@ -9,6 +9,8 @@ class Reader {
 public:
     virtual void read(uint8_t* data, uint64_t size) = 0;
     virtual ~Reader() = default;
+
+    virtual bool finished() = 0;
 };
 
 } // namespace common

src/include/main/client_context.h

@@ -1,17 +1,14 @@
 #pragma once
 
 #include <atomic>
-#include <cstdint>
 #include <memory>
 #include <mutex>
 
 #include "client_config.h"
 #include "common/task_system/progress_bar.h"
 #include "common/timer.h"
 #include "common/types/value/value.h"
-#include "function/function.h"
 #include "function/table/scan_replacement.h"
-#include "main/kuzu_fwd.h"
 #include "parser/statement.h"
 #include "prepared_statement.h"
 #include "query_result.h"

src/include/storage/stats/table_statistics_collection.h

@@ -38,6 +38,7 @@ class TablesStatistics {
 
     inline void checkpointInMemoryIfNecessary() {
         std::unique_lock lck{mtx};
+        KU_ASSERT(readWriteVersion);
         readOnlyVersion = std::move(readWriteVersion);
         resetToNotUpdated();
     }

src/include/storage/storage_structure/db_file_utils.h

@@ -1,17 +1,20 @@
 #pragma once
 
-#include <cstdint>
 #include <functional>
 
 #include "common/types/types.h"
-#include "storage/buffer_manager/bm_file_handle.h"
-#include "storage/buffer_manager/buffer_manager.h"
-#include "storage/wal/wal.h"
-#include "transaction/transaction.h"
 
 namespace kuzu {
+namespace transaction {
+enum class TransactionType : uint8_t;
+} // namespace transaction
+
 namespace storage {
 
+struct DBFileID;
+class BMFileHandle;
+class BufferManager;
+class WAL;
 class DBFileUtils {
 public:
     constexpr static common::page_idx_t NULL_PAGE_IDX = common::INVALID_PAGE_IDX;