From 842bff15f21dfcb957c20b19eafb6de4bbfe37ca Mon Sep 17 00:00:00 2001 From: Guodong Jin Date: Sun, 11 Aug 2024 13:19:08 -0400 Subject: [PATCH] rework checkpoint mem estimation (#4055) add debug prefix to enable_multi_writes (cherry picked from commit 6a6517854c1bdf1e359ce64fb387efcb5abd2e33) --- src/include/main/db_config.h | 2 +- src/include/main/settings.h | 103 +++++++++--------- src/include/storage/index/in_mem_hash_index.h | 1 + .../storage/local_storage/local_hash_index.h | 8 ++ .../storage/local_storage/local_node_table.h | 1 + .../storage/local_storage/local_rel_table.h | 1 + .../storage/local_storage/local_storage.h | 2 + .../storage/local_storage/local_table.h | 1 + src/include/storage/storage_manager.h | 2 - .../storage/storage_structure/disk_array.h | 6 + src/include/storage/store/node_table.h | 2 - src/include/storage/store/rel_table.h | 2 - src/include/storage/store/table.h | 2 - src/include/storage/undo_buffer.h | 2 + src/include/transaction/transaction.h | 1 + src/main/database.cpp | 8 +- .../local_storage/local_node_table.cpp | 4 + src/storage/local_storage/local_rel_table.cpp | 9 ++ src/storage/local_storage/local_storage.cpp | 8 ++ src/storage/storage_manager.cpp | 9 -- src/storage/store/node_table.cpp | 4 - src/storage/undo_buffer.cpp | 8 ++ src/transaction/transaction.cpp | 4 + src/transaction/transaction_manager.cpp | 3 +- test/include/graph_test/base_graph_test.h | 5 +- test/runner/e2e_test.cpp | 4 +- .../concurrency/ddl_concurrent_execution.test | 4 +- .../concurrency/ddl_serial_execution.test | 12 +- .../dml_empty_serial_execution.test | 26 ++--- 29 files changed, 143 insertions(+), 101 deletions(-) diff --git a/src/include/main/db_config.h b/src/include/main/db_config.h index 4e9c794dcc9..2e18f76b279 100644 --- a/src/include/main/db_config.h +++ b/src/include/main/db_config.h @@ -16,7 +16,7 @@ class ClientContext; struct SystemConfig; typedef void (*set_context)(ClientContext* context, const common::Value& parameter); -typedef common::Value (*get_setting)(ClientContext* context); +typedef common::Value (*get_setting)(const ClientContext* context); enum class OptionType : uint8_t { CONFIGURATION = 0, EXTENSION = 1 }; diff --git a/src/include/main/settings.h b/src/include/main/settings.h index 7f153bcdb33..bd2248ebcfc 100644 --- a/src/include/main/settings.h +++ b/src/include/main/settings.h @@ -1,5 +1,6 @@ #pragma once +#include "common/exception/not_implemented.h" #include "common/types/value/value.h" #include "main/client_context.h" #include "main/db_config.h" @@ -8,202 +9,204 @@ namespace kuzu { namespace main { struct ThreadsSetting { - static constexpr const char* name = "threads"; - static constexpr const common::LogicalTypeID inputType = common::LogicalTypeID::INT64; + static constexpr auto name = "threads"; + static constexpr auto inputType = common::LogicalTypeID::INT64; static void setContext(ClientContext* context, const common::Value& parameter) { parameter.validateType(inputType); context->getClientConfigUnsafe()->numThreads = parameter.getValue(); } - static common::Value getSetting(ClientContext* context) { + static common::Value getSetting(const ClientContext* context) { return common::Value(context->getClientConfig()->numThreads); } }; struct TimeoutSetting { - static constexpr const char* name = "timeout"; - static constexpr const common::LogicalTypeID inputType = common::LogicalTypeID::INT64; + static constexpr auto name = "timeout"; + static constexpr auto inputType = common::LogicalTypeID::INT64; static void setContext(ClientContext* context, const common::Value& parameter) { parameter.validateType(inputType); context->getClientConfigUnsafe()->timeoutInMS = parameter.getValue(); } - static common::Value getSetting(ClientContext* context) { + static common::Value getSetting(const ClientContext* context) { return common::Value(context->getClientConfig()->timeoutInMS); } }; struct ProgressBarSetting { - static constexpr const char* name = "progress_bar"; - static constexpr const common::LogicalTypeID inputType = common::LogicalTypeID::BOOL; + static constexpr auto name = "progress_bar"; + static constexpr auto inputType = common::LogicalTypeID::BOOL; static void setContext(ClientContext* context, const common::Value& parameter) { parameter.validateType(inputType); context->getClientConfigUnsafe()->enableProgressBar = parameter.getValue(); context->getProgressBar()->toggleProgressBarPrinting(parameter.getValue()); } - static common::Value getSetting(ClientContext* context) { + static common::Value getSetting(const ClientContext* context) { return common::Value(context->getClientConfig()->enableProgressBar); } }; struct ProgressBarTimerSetting { - static constexpr const char* name = "progress_bar_time"; - static constexpr const common::LogicalTypeID inputType = common::LogicalTypeID::INT64; + static constexpr auto name = "progress_bar_time"; + static constexpr auto inputType = common::LogicalTypeID::INT64; static void setContext(ClientContext* context, const common::Value& parameter) { parameter.validateType(inputType); context->getClientConfigUnsafe()->showProgressAfter = parameter.getValue(); context->getProgressBar()->setShowProgressAfter(parameter.getValue()); } - static common::Value getSetting(ClientContext* context) { + static common::Value getSetting(const ClientContext* context) { return common::Value(context->getClientConfig()->showProgressAfter); } }; struct VarLengthExtendMaxDepthSetting { - static constexpr const char* name = "var_length_extend_max_depth"; - static constexpr const common::LogicalTypeID inputType = common::LogicalTypeID::INT64; + static constexpr auto name = "var_length_extend_max_depth"; + static constexpr auto inputType = common::LogicalTypeID::INT64; static void setContext(ClientContext* context, const common::Value& parameter) { parameter.validateType(inputType); context->getClientConfigUnsafe()->varLengthMaxDepth = parameter.getValue(); } - static common::Value getSetting(ClientContext* context) { + static common::Value getSetting(const ClientContext* context) { return common::Value(context->getClientConfig()->varLengthMaxDepth); } }; struct EnableSemiMaskSetting { - static constexpr const char* name = "enable_semi_mask"; - static constexpr const common::LogicalTypeID inputType = common::LogicalTypeID::BOOL; + static constexpr auto name = "enable_semi_mask"; + static constexpr auto inputType = common::LogicalTypeID::BOOL; static void setContext(ClientContext* context, const common::Value& parameter) { parameter.validateType(inputType); context->getClientConfigUnsafe()->enableSemiMask = parameter.getValue(); } - static common::Value getSetting(ClientContext* context) { + static common::Value getSetting(const ClientContext* context) { return common::Value(context->getClientConfig()->enableSemiMask); } }; struct DisableMapKeyCheck { - static constexpr const char* name = "disable_map_key_check"; - static constexpr const common::LogicalTypeID inputType = common::LogicalTypeID::BOOL; + static constexpr auto name = "disable_map_key_check"; + static constexpr auto inputType = common::LogicalTypeID::BOOL; static void setContext(ClientContext* context, const common::Value& parameter) { parameter.validateType(inputType); context->getClientConfigUnsafe()->disableMapKeyCheck = parameter.getValue(); } - static common::Value getSetting(ClientContext* context) { + static common::Value getSetting(const ClientContext* context) { return common::Value(context->getClientConfig()->disableMapKeyCheck); } }; struct EnableZoneMapSetting { - static constexpr const char* name = "enable_zone_map"; - static constexpr const common::LogicalTypeID inputType = common::LogicalTypeID::BOOL; + static constexpr auto name = "enable_zone_map"; + static constexpr auto inputType = common::LogicalTypeID::BOOL; static void setContext(ClientContext* /*context*/, const common::Value& parameter) { parameter.validateType(inputType); + // TODO(Guodong/Xiyang/Ben): Turn me on when zone map is ready. + throw common::NotImplementedException("Zone map is not yet ready to be turned on."); // context->getClientConfigUnsafe()->enableZoneMap = parameter.getValue(); } - static common::Value getSetting(ClientContext* context) { + static common::Value getSetting(const ClientContext* context) { return common::Value(context->getClientConfig()->enableZoneMap); } }; struct HomeDirectorySetting { - static constexpr const char* name = "home_directory"; - static constexpr const common::LogicalTypeID inputType = common::LogicalTypeID::STRING; + static constexpr auto name = "home_directory"; + static constexpr auto inputType = common::LogicalTypeID::STRING; static void setContext(ClientContext* context, const common::Value& parameter) { parameter.validateType(inputType); context->getClientConfigUnsafe()->homeDirectory = parameter.getValue(); } - static common::Value getSetting(ClientContext* context) { + static common::Value getSetting(const ClientContext* context) { return common::Value::createValue(context->getClientConfig()->homeDirectory); } }; struct FileSearchPathSetting { - static constexpr const char* name = "file_search_path"; - static constexpr const common::LogicalTypeID inputType = common::LogicalTypeID::STRING; + static constexpr auto name = "file_search_path"; + static constexpr auto inputType = common::LogicalTypeID::STRING; static void setContext(ClientContext* context, const common::Value& parameter) { parameter.validateType(inputType); context->getClientConfigUnsafe()->fileSearchPath = parameter.getValue(); } - static common::Value getSetting(ClientContext* context) { + static common::Value getSetting(const ClientContext* context) { return common::Value::createValue(context->getClientConfig()->fileSearchPath); } }; struct RecursivePatternSemanticSetting { - static constexpr const char* name = "recursive_pattern_semantic"; - static constexpr const common::LogicalTypeID inputType = common::LogicalTypeID::STRING; + static constexpr auto name = "recursive_pattern_semantic"; + static constexpr auto inputType = common::LogicalTypeID::STRING; static void setContext(ClientContext* context, const common::Value& parameter) { parameter.validateType(inputType); - auto input = parameter.getValue(); + const auto input = parameter.getValue(); context->getClientConfigUnsafe()->recursivePatternSemantic = common::PathSemanticUtils::fromString(input); } - static common::Value getSetting(ClientContext* context) { - auto result = common::PathSemanticUtils::toString( + static common::Value getSetting(const ClientContext* context) { + const auto result = common::PathSemanticUtils::toString( context->getClientConfig()->recursivePatternSemantic); return common::Value::createValue(result); } }; struct RecursivePatternFactorSetting { - static constexpr const char* name = "recursive_pattern_factor"; - static constexpr const common::LogicalTypeID inputType = common::LogicalTypeID::INT64; + static constexpr auto name = "recursive_pattern_factor"; + static constexpr auto inputType = common::LogicalTypeID::INT64; static void setContext(ClientContext* context, const common::Value& parameter) { parameter.validateType(inputType); context->getClientConfigUnsafe()->recursivePatternCardinalityScaleFactor = parameter.getValue(); } - static common::Value getSetting(ClientContext* context) { + static common::Value getSetting(const ClientContext* context) { return common::Value::createValue( context->getClientConfig()->recursivePatternCardinalityScaleFactor); } }; struct EnableMVCCSetting { - static constexpr const char* name = "enable_multi_writes"; - static constexpr const common::LogicalTypeID inputType = common::LogicalTypeID::BOOL; + static constexpr auto name = "debug_enable_multi_writes"; + static constexpr auto inputType = common::LogicalTypeID::BOOL; static void setContext(ClientContext* context, const common::Value& parameter) { KU_ASSERT(parameter.getDataType().getLogicalTypeID() == common::LogicalTypeID::BOOL); // TODO: This is a temporary solution to make tests of multiple write transactions easier. context->getDBConfigUnsafe()->enableMultiWrites = parameter.getValue(); } - static common::Value getSetting(ClientContext* context) { + static common::Value getSetting(const ClientContext* context) { return common::Value(context->getDBConfig()->enableMultiWrites); } }; struct CheckpointThresholdSetting { - static constexpr const char* name = "checkpoint_threshold"; - static constexpr common::LogicalTypeID inputType = common::LogicalTypeID::INT64; + static constexpr auto name = "checkpoint_threshold"; + static constexpr auto inputType = common::LogicalTypeID::INT64; static void setContext(ClientContext* context, const common::Value& parameter) { parameter.validateType(inputType); context->getDBConfigUnsafe()->checkpointThreshold = parameter.getValue(); } - static common::Value getSetting(ClientContext* context) { + static common::Value getSetting(const ClientContext* context) { return common::Value(context->getDBConfig()->checkpointThreshold); } }; struct AutoCheckpointSetting { - static constexpr const char* name = "auto_checkpoint"; - static constexpr common::LogicalTypeID inputType = common::LogicalTypeID::BOOL; + static constexpr auto name = "auto_checkpoint"; + static constexpr auto inputType = common::LogicalTypeID::BOOL; static void setContext(ClientContext* context, const common::Value& parameter) { parameter.validateType(inputType); context->getDBConfigUnsafe()->autoCheckpoint = parameter.getValue(); } - static common::Value getSetting(ClientContext* context) { + static common::Value getSetting(const ClientContext* context) { return common::Value(context->getDBConfig()->autoCheckpoint); } }; struct ForceCheckpointClosingDBSetting { - static constexpr const char* name = "force_checkpoint_on_close"; - static constexpr common::LogicalTypeID inputType = common::LogicalTypeID::BOOL; + static constexpr auto name = "force_checkpoint_on_close"; + static constexpr auto inputType = common::LogicalTypeID::BOOL; static void setContext(ClientContext* context, const common::Value& parameter) { parameter.validateType(inputType); context->getDBConfigUnsafe()->forceCheckpointOnClose = parameter.getValue(); } - static common::Value getSetting(ClientContext* context) { + static common::Value getSetting(const ClientContext* context) { return common::Value(context->getDBConfig()->forceCheckpointOnClose); } }; diff --git a/src/include/storage/index/in_mem_hash_index.h b/src/include/storage/index/in_mem_hash_index.h index 2cbf22a4391..a433b19e7cb 100644 --- a/src/include/storage/index/in_mem_hash_index.h +++ b/src/include/storage/index/in_mem_hash_index.h @@ -145,6 +145,7 @@ class InMemHashIndex final { uint64_t numPrimarySlots() const { return pSlots->size(); } uint64_t numOverflowSlots() const { return oSlots->size(); } + uint64_t getEstimatedMemUsage() const { return pSlots->getMemUsage() + oSlots->getMemUsage(); } const HashIndexHeader& getIndexHeader() const { return indexHeader; } diff --git a/src/include/storage/local_storage/local_hash_index.h b/src/include/storage/local_storage/local_hash_index.h index fd53f0fbaf4..bcb736908d4 100644 --- a/src/include/storage/local_storage/local_hash_index.h +++ b/src/include/storage/local_storage/local_hash_index.h @@ -12,6 +12,7 @@ namespace storage { class BaseHashIndexLocalStorage { public: virtual ~BaseHashIndexLocalStorage() = default; + virtual uint64_t getEstimatedMemUsage() = 0; }; enum class HashIndexLocalLookupState : uint8_t { KEY_FOUND, KEY_DELETED, KEY_NOT_EXIST }; @@ -78,6 +79,11 @@ class HashIndexLocalStorage final : public BaseHashIndexLocalStorage { const InMemHashIndex& getInsertions() { return localInsertions; } + uint64_t getEstimatedMemUsage() override { + return localInsertions.getEstimatedMemUsage() + + localDeletions.size() * sizeof(OwnedKeyType); + } + private: // When the storage type is string, allow the key type to be string_view with a custom hash // function @@ -176,6 +182,8 @@ class LocalHashIndex { ->deleteKey(key); } + uint64_t getEstimatedMemUsage() { return localIndex->getEstimatedMemUsage(); } + private: common::PhysicalTypeID keyDataTypeID; std::unique_ptr localIndex; diff --git a/src/include/storage/local_storage/local_node_table.h b/src/include/storage/local_storage/local_node_table.h index a472f267118..4241577a675 100644 --- a/src/include/storage/local_storage/local_node_table.h +++ b/src/include/storage/local_storage/local_node_table.h @@ -22,6 +22,7 @@ class LocalNodeTable final : public LocalTable { bool delete_(transaction::Transaction* transaction, TableDeleteState& deleteState) override; bool addColumn(transaction::Transaction* transaction, TableAddColumnState& addColumnState) override; + uint64_t getEstimatedMemUsage() override; common::offset_t validateUniquenessConstraint(const transaction::Transaction* transaction, const common::ValueVector& pkVector); diff --git a/src/include/storage/local_storage/local_rel_table.h b/src/include/storage/local_storage/local_rel_table.h index e08c4adb227..bc7a94d7a26 100644 --- a/src/include/storage/local_storage/local_rel_table.h +++ b/src/include/storage/local_storage/local_rel_table.h @@ -26,6 +26,7 @@ class LocalRelTable final : public LocalTable { bool delete_(transaction::Transaction* transaction, TableDeleteState& state) override; bool addColumn(transaction::Transaction* transaction, TableAddColumnState& addColumnState) override; + uint64_t getEstimatedMemUsage() override; void checkIfNodeHasRels(common::ValueVector* srcNodeIDVector) const; diff --git a/src/include/storage/local_storage/local_storage.h b/src/include/storage/local_storage/local_storage.h index 32f73853412..9a4d808db00 100644 --- a/src/include/storage/local_storage/local_storage.h +++ b/src/include/storage/local_storage/local_storage.h @@ -29,6 +29,8 @@ class LocalStorage { void commit(); void rollback(); + uint64_t getEstimatedMemUsage() const; + private: main::ClientContext& clientContext; std::unordered_map> tables; diff --git a/src/include/storage/local_storage/local_table.h b/src/include/storage/local_storage/local_table.h index a9fec570f89..3aa6e5fd21a 100644 --- a/src/include/storage/local_storage/local_table.h +++ b/src/include/storage/local_storage/local_table.h @@ -33,6 +33,7 @@ class LocalTable { TableAddColumnState& addColumnState) = 0; virtual void clear() = 0; virtual common::TableType getTableType() const = 0; + virtual uint64_t getEstimatedMemUsage() = 0; template const TARGET& constCast() { diff --git a/src/include/storage/storage_manager.h b/src/include/storage/storage_manager.h index 2c244fd3b5c..364d859e5e7 100644 --- a/src/include/storage/storage_manager.h +++ b/src/include/storage/storage_manager.h @@ -45,8 +45,6 @@ class StorageManager { bool isReadOnly() const { return readOnly; } bool compressionEnabled() const { return enableCompression; } - uint64_t getEstimatedMemoryUsage(); - private: BMFileHandle* initFileHandle(const std::string& fileName, common::VirtualFileSystem* vfs, main::ClientContext* context) const; diff --git a/src/include/storage/storage_structure/disk_array.h b/src/include/storage/storage_structure/disk_array.h index 302c8680aee..126e6fc04da 100644 --- a/src/include/storage/storage_structure/disk_array.h +++ b/src/include/storage/storage_structure/disk_array.h @@ -357,6 +357,10 @@ class BlockVectorInternal { // memory and not on disk (nor on the wal). uint8_t* operator[](uint64_t idx); + uint64_t getMemUsage() const { + return inMemArrayPages.size() * common::BufferPoolConstants::PAGE_4KB_SIZE; + } + protected: inline uint64_t addInMemoryArrayPage(bool setToZero) { inMemArrayPages.emplace_back( @@ -399,6 +403,8 @@ class BlockVector { return DiskArray::getAlignedElementSize(); } + uint64_t getMemUsage() const { return vector.getMemUsage(); } + private: BlockVectorInternal vector; }; diff --git a/src/include/storage/store/node_table.h b/src/include/storage/store/node_table.h index a5cb0c59df9..f12c36f8a09 100644 --- a/src/include/storage/store/node_table.h +++ b/src/include/storage/store/node_table.h @@ -153,8 +153,6 @@ class NodeTable final : public Table { void rollback(LocalTable* localTable) override; void checkpoint(common::Serializer& ser) override; - uint64_t getEstimatedMemoryUsage() const override; - common::node_group_idx_t getNumCommittedNodeGroups() const { return nodeGroups->getNumNodeGroups(); } diff --git a/src/include/storage/store/rel_table.h b/src/include/storage/store/rel_table.h index 2b22e949b5c..5bdfc00d96e 100644 --- a/src/include/storage/store/rel_table.h +++ b/src/include/storage/store/rel_table.h @@ -168,8 +168,6 @@ class RelTable final : public Table { void rollback(LocalTable* localTable) override; void checkpoint(common::Serializer& ser) override; - uint64_t getEstimatedMemoryUsage() const override { return 0; } - common::row_idx_t getNumRows() override { return nextRelOffset; } RelTableData* getDirectedTableData(common::RelDataDirection direction) const { diff --git a/src/include/storage/store/table.h b/src/include/storage/store/table.h index 8f239657d4d..23dcd51efcb 100644 --- a/src/include/storage/store/table.h +++ b/src/include/storage/store/table.h @@ -167,8 +167,6 @@ class Table { virtual void rollback(LocalTable* localTable) = 0; virtual void checkpoint(common::Serializer& ser) = 0; - virtual uint64_t getEstimatedMemoryUsage() const = 0; - virtual common::row_idx_t getNumRows() = 0; void setHasChanges() { hasChanges = true; } diff --git a/src/include/storage/undo_buffer.h b/src/include/storage/undo_buffer.h index e6d7ac4e2b3..dae6cefa539 100644 --- a/src/include/storage/undo_buffer.h +++ b/src/include/storage/undo_buffer.h @@ -95,6 +95,8 @@ class UndoBuffer { void commit(common::transaction_t commitTS) const; void rollback(); + uint64_t getMemUsage() const; + private: uint8_t* createUndoRecord(uint64_t size); diff --git a/src/include/transaction/transaction.h b/src/include/transaction/transaction.h index 31765d5f57a..f9a49e8dfcb 100644 --- a/src/include/transaction/transaction.h +++ b/src/include/transaction/transaction.h @@ -77,6 +77,7 @@ class Transaction { void commit(storage::WAL* wal) const; void rollback(storage::WAL* wal) const; + uint64_t getEstimatedMemUsage() const; storage::LocalStorage* getLocalStorage() const { return localStorage.get(); } bool hasNewlyInsertedNodes(common::table_id_t tableID) const { return maxCommittedNodeOffsets.contains(tableID); diff --git a/src/main/database.cpp b/src/main/database.cpp index ddd4428c3df..7ccb162b27d 100644 --- a/src/main/database.cpp +++ b/src/main/database.cpp @@ -97,9 +97,11 @@ Database::Database(std::string_view databasePath, SystemConfig systemConfig) } Database::~Database() { - if (dbConfig.forceCheckpointOnClose) { - ClientContext clientContext(this); - transactionManager->checkpoint(clientContext); + if (!dbConfig.readOnly && dbConfig.forceCheckpointOnClose) { + try { + ClientContext clientContext(this); + transactionManager->checkpoint(clientContext); + } catch (...) {} // NOLINT } } diff --git a/src/storage/local_storage/local_node_table.cpp b/src/storage/local_storage/local_node_table.cpp index 7935c418290..5663fce89bf 100644 --- a/src/storage/local_storage/local_node_table.cpp +++ b/src/storage/local_storage/local_node_table.cpp @@ -101,6 +101,10 @@ bool LocalNodeTable::addColumn(Transaction* transaction, TableAddColumnState& ad return true; } +uint64_t LocalNodeTable::getEstimatedMemUsage() { + return nodeGroups.getEstimatedMemoryUsage() + hashIndex->getEstimatedMemUsage(); +} + void LocalNodeTable::clear() { auto& nodeTable = ku_dynamic_cast(table); hashIndex = std::make_unique( diff --git a/src/storage/local_storage/local_rel_table.cpp b/src/storage/local_storage/local_rel_table.cpp index ff41f981f2a..d760c0c463c 100644 --- a/src/storage/local_storage/local_rel_table.cpp +++ b/src/storage/local_storage/local_rel_table.cpp @@ -125,6 +125,15 @@ bool LocalRelTable::addColumn(Transaction* transaction, TableAddColumnState& add return true; } +uint64_t LocalRelTable::getEstimatedMemUsage() { + // Esimation of fwdIndex and bwdIndex is rough. + if (!localNodeGroup) { + return 0; + } + return localNodeGroup->getEstimatedMemoryUsage() + fwdIndex.size() * sizeof(offset_t) + + bwdIndex.size() * sizeof(offset_t); +} + void LocalRelTable::checkIfNodeHasRels(ValueVector* srcNodeIDVector) const { KU_ASSERT(srcNodeIDVector->state->isFlat()); const auto nodeIDPos = srcNodeIDVector->state->getSelVector()[0]; diff --git a/src/storage/local_storage/local_storage.cpp b/src/storage/local_storage/local_storage.cpp index aee80504a33..79ab9fcb787 100644 --- a/src/storage/local_storage/local_storage.cpp +++ b/src/storage/local_storage/local_storage.cpp @@ -61,5 +61,13 @@ void LocalStorage::rollback() { } } +uint64_t LocalStorage::getEstimatedMemUsage() const { + uint64_t totalMemUsage = 0; + for (const auto& [tableID, localTable] : tables) { + totalMemUsage += localTable->getEstimatedMemUsage(); + } + return totalMemUsage; +} + } // namespace storage } // namespace kuzu diff --git a/src/storage/storage_manager.cpp b/src/storage/storage_manager.cpp index b049b4ec42a..8ed0414247d 100644 --- a/src/storage/storage_manager.cpp +++ b/src/storage/storage_manager.cpp @@ -207,15 +207,6 @@ ShadowFile& StorageManager::getShadowFile() const { return *shadowFile; } -uint64_t StorageManager::getEstimatedMemoryUsage() { - std::lock_guard lck{mtx}; - uint64_t totalMemoryUsage = 0; - for (const auto& [tableID, table] : tables) { - totalMemoryUsage += table->getEstimatedMemoryUsage(); - } - return totalMemoryUsage; -} - void StorageManager::checkpoint(main::ClientContext& clientContext) { if (main::DBConfig::isDBPathInMemory(databasePath)) { return; diff --git a/src/storage/store/node_table.cpp b/src/storage/store/node_table.cpp index 912d78adb01..5d5ae306112 100644 --- a/src/storage/store/node_table.cpp +++ b/src/storage/store/node_table.cpp @@ -399,10 +399,6 @@ void NodeTable::serialize(Serializer& serializer) const { nodeGroups->serialize(serializer); } -uint64_t NodeTable::getEstimatedMemoryUsage() const { - return nodeGroups->getEstimatedMemoryUsage(); -} - bool NodeTable::isVisible(const Transaction* transaction, offset_t offset) const { auto [nodeGroupIdx, offsetInGroup] = StorageUtils::getNodeGroupIdxAndOffsetInChunk(offset); auto* nodeGroup = getNodeGroupNoLock(nodeGroupIdx); diff --git a/src/storage/undo_buffer.cpp b/src/storage/undo_buffer.cpp index 1ffc4e5432a..7a9fa6f5fcd 100644 --- a/src/storage/undo_buffer.cpp +++ b/src/storage/undo_buffer.cpp @@ -172,6 +172,14 @@ void UndoBuffer::rollback() { [&](UndoRecordType entryType, uint8_t const* entry) { rollbackRecord(entryType, entry); }); } +uint64_t UndoBuffer::getMemUsage() const { + uint64_t totalMemUsage = 0; + for (const auto& buffer : memoryBuffers) { + totalMemUsage += buffer.getSize(); + } + return totalMemUsage; +} + void UndoBuffer::commitRecord(UndoRecordType recordType, const uint8_t* record, transaction_t commitTS) const { switch (recordType) { diff --git a/src/transaction/transaction.cpp b/src/transaction/transaction.cpp index 20e1f18df30..570cbf1ff6b 100644 --- a/src/transaction/transaction.cpp +++ b/src/transaction/transaction.cpp @@ -51,6 +51,10 @@ void Transaction::rollback(storage::WAL* wal) const { } } +uint64_t Transaction::getEstimatedMemUsage() const { + return localStorage->getEstimatedMemUsage() + undoBuffer->getMemUsage(); +} + void Transaction::pushCatalogEntry(CatalogSet& catalogSet, CatalogEntry& catalogEntry, bool skipLoggingToWAL) const { undoBuffer->createCatalogEntry(catalogSet, catalogEntry); diff --git a/src/transaction/transaction_manager.cpp b/src/transaction/transaction_manager.cpp index d2d4e8b167a..0ea1e76a24a 100644 --- a/src/transaction/transaction_manager.cpp +++ b/src/transaction/transaction_manager.cpp @@ -139,8 +139,7 @@ bool TransactionManager::canAutoCheckpoint(const main::ClientContext& clientCont // Recovery transactions are not allowed to trigger auto checkpoint. return false; } - const auto expectedSize = - clientContext.getStorageManager()->getEstimatedMemoryUsage() + wal.getFileSize(); + const auto expectedSize = clientContext.getTx()->getEstimatedMemUsage() + wal.getFileSize(); return expectedSize > clientContext.getDBConfig()->checkpointThreshold; } diff --git a/test/include/graph_test/base_graph_test.h b/test/include/graph_test/base_graph_test.h index 45e60f84a95..c6d58c5d186 100644 --- a/test/include/graph_test/base_graph_test.h +++ b/test/include/graph_test/base_graph_test.h @@ -35,8 +35,11 @@ class BaseGraphTest : public Test { virtual std::string getInputDir() = 0; void TearDown() override { + conn.reset(); + connMap.clear(); + database.reset(); if (!inMemMode) { - removeDir(databasePath); + std::filesystem::remove_all(databasePath); } } diff --git a/test/runner/e2e_test.cpp b/test/runner/e2e_test.cpp index 3cdc82688f7..a5a08d20a89 100644 --- a/test/runner/e2e_test.cpp +++ b/test/runner/e2e_test.cpp @@ -65,8 +65,8 @@ class EndToEndTest : public DBTest { } void TearDown() override { - std::filesystem::remove_all(databasePath); - BaseGraphTest::removeIEDBPath(); + DBTest::TearDown(); + removeIEDBPath(); if (datasetType == TestGroup::DatasetType::CSV_TO_PARQUET) { std::filesystem::remove_all(parquetTempDatasetPath); } diff --git a/test/test_files/transaction/concurrency/ddl_concurrent_execution.test b/test/test_files/transaction/concurrency/ddl_concurrent_execution.test index 891cfcc7a4c..04131e38759 100644 --- a/test/test_files/transaction/concurrency/ddl_concurrent_execution.test +++ b/test/test_files/transaction/concurrency/ddl_concurrent_execution.test @@ -54,7 +54,7 @@ 16|propA|INT64|False -CASE ConcurrentMultiWriteCreateTable --STATEMENT CALL ENABLE_MULTI_WRITES=true; +-STATEMENT CALL debug_enable_multi_writes=true; ---- ok -CREATE_CONNECTION conn2 -CREATE_CONNECTION conn3 @@ -73,7 +73,7 @@ #TODO(Sam/Guodong): FIX-ME. -CASE ConcurrentMultiWriteAlter -SKIP --STATEMENT CALL ENABLE_MULTI_WRITES=true; +-STATEMENT CALL debug_enable_multi_writes=true; ---- ok -CREATE_CONNECTION conn2 -CREATE_CONNECTION conn3 diff --git a/test/test_files/transaction/concurrency/ddl_serial_execution.test b/test/test_files/transaction/concurrency/ddl_serial_execution.test index 18a2bec90d5..c7f97a14b82 100644 --- a/test/test_files/transaction/concurrency/ddl_serial_execution.test +++ b/test/test_files/transaction/concurrency/ddl_serial_execution.test @@ -17,9 +17,9 @@ Cannot start a new write transaction in the system. Only one write transaction a 0|a|INT32|True 1|b|INT32|False -# Test CALL ENABLE_MULTI_WRITES=true +# Test CALL debug_enable_multi_writes=true -CASE EnableMultiWrites --STATEMENT CALL ENABLE_MULTI_WRITES=true; +-STATEMENT CALL debug_enable_multi_writes=true; ---- ok -STATEMENT BEGIN TRANSACTION; ---- ok @@ -77,7 +77,7 @@ Catalog exception: t does not exist in catalog. # Test W-W conflict -CASE CreateTableWriteWriteConflict --STATEMENT CALL ENABLE_MULTI_WRITES=true; +-STATEMENT CALL debug_enable_multi_writes=true; ---- ok -STATEMENT BEGIN TRANSACTION; ---- ok @@ -158,7 +158,7 @@ Catalog exception: t does not exist in catalog. 1|b|INT32|False -CASE CreateAndDropMultiConnections --STATEMENT CALL ENABLE_MULTI_WRITES=true; +-STATEMENT CALL debug_enable_multi_writes=true; ---- ok -STATEMENT BEGIN TRANSACTION; ---- ok @@ -206,7 +206,7 @@ Binder exception: Table t does not exist. 1|b|INT32|False -CASE RenameTable --STATEMENT CALL ENABLE_MULTI_WRITES=true; +-STATEMENT CALL debug_enable_multi_writes=true; ---- ok -STATEMENT BEGIN TRANSACTION; ---- ok @@ -243,7 +243,7 @@ Catalog exception: t does not exist in catalog. 1|b|INT32|False -CASE RenameColumn --STATEMENT CALL ENABLE_MULTI_WRITES=true; +-STATEMENT CALL debug_enable_multi_writes=true; ---- ok -STATEMENT CALL auto_checkpoint=false; ---- ok diff --git a/test/test_files/transaction/concurrency/dml_empty_serial_execution.test b/test/test_files/transaction/concurrency/dml_empty_serial_execution.test index 877be09893f..149fc7594e8 100644 --- a/test/test_files/transaction/concurrency/dml_empty_serial_execution.test +++ b/test/test_files/transaction/concurrency/dml_empty_serial_execution.test @@ -20,7 +20,7 @@ ] -CASE WWConflictNodeCopyUpdate --STATEMENT CALL ENABLE_MULTI_WRITES=true; +-STATEMENT CALL debug_enable_multi_writes=true; ---- ok -INSERT_STATEMENT_BLOCK COPY_TINYSNB_PERSON -CREATE_CONNECTION conn2 @@ -35,7 +35,7 @@ Runtime exception: Write-write conflict of updating the same row. -CASE WWConflictNodeCopyDelete --STATEMENT CALL ENABLE_MULTI_WRITES=true; +-STATEMENT CALL debug_enable_multi_writes=true; ---- ok -INSERT_STATEMENT_BLOCK COPY_TINYSNB_PERSON -CREATE_CONNECTION conn2 @@ -50,7 +50,7 @@ Runtime exception: Write-write conflict of updating the same row. Runtime exception: Write-write conflict: deleting a row that is already deleted by another transaction. -CASE WWConflictRelCopyUpdate --STATEMENT CALL ENABLE_MULTI_WRITES=true; +-STATEMENT CALL debug_enable_multi_writes=true; ---- ok -INSERT_STATEMENT_BLOCK COPY_TINYSNB_PERSON -INSERT_STATEMENT_BLOCK COPY_TINYSNB_KNOWS @@ -66,7 +66,7 @@ Runtime exception: Write-write conflict: deleting a row that is already deleted Runtime exception: Write-write conflict of updating the same row. -CASE WWConflictRelCopyDelete --STATEMENT CALL ENABLE_MULTI_WRITES=true; +-STATEMENT CALL debug_enable_multi_writes=true; ---- ok -INSERT_STATEMENT_BLOCK COPY_TINYSNB_PERSON -INSERT_STATEMENT_BLOCK COPY_TINYSNB_KNOWS @@ -82,7 +82,7 @@ Runtime exception: Write-write conflict of updating the same row. Runtime exception: Write-write conflict: deleting a row that is already deleted by another transaction. -CASE WWConflictNodeInsertUpdate --STATEMENT CALL ENABLE_MULTI_WRITES=true; +-STATEMENT CALL debug_enable_multi_writes=true; ---- ok -CREATE_DATASET_SCHEMA tinysnb -INSERT_DATASET_BY_ROW tinysnb @@ -98,7 +98,7 @@ Runtime exception: Write-write conflict: deleting a row that is already deleted Runtime exception: Write-write conflict of updating the same row. -CASE WWConflictNodeInsertDelete --STATEMENT CALL ENABLE_MULTI_WRITES=true; +-STATEMENT CALL debug_enable_multi_writes=true; ---- ok -CREATE_DATASET_SCHEMA tinysnb -INSERT_DATASET_BY_ROW tinysnb @@ -114,7 +114,7 @@ Runtime exception: Write-write conflict of updating the same row. Runtime exception: Write-write conflict: deleting a row that is already deleted by another transaction. -CASE WWConflictRelInsertUpdate --STATEMENT CALL ENABLE_MULTI_WRITES=true; +-STATEMENT CALL debug_enable_multi_writes=true; ---- ok -CREATE_DATASET_SCHEMA tinysnb -INSERT_DATASET_BY_ROW tinysnb @@ -130,7 +130,7 @@ Runtime exception: Write-write conflict: deleting a row that is already deleted Runtime exception: Write-write conflict of updating the same row. -CASE WWConflictRelInsertDelete --STATEMENT CALL ENABLE_MULTI_WRITES=true; +-STATEMENT CALL debug_enable_multi_writes=true; ---- ok -CREATE_DATASET_SCHEMA tinysnb -INSERT_DATASET_BY_ROW tinysnb @@ -146,7 +146,7 @@ Runtime exception: Write-write conflict of updating the same row. Runtime exception: Write-write conflict: deleting a row that is already deleted by another transaction. -CASE MultiTransactionNodeInsert --STATEMENT CALL ENABLE_MULTI_WRITES=true; +-STATEMENT CALL debug_enable_multi_writes=true; ---- ok -STATEMENT CALL auto_checkpoint=false; ---- ok @@ -200,7 +200,7 @@ Runtime exception: Write-write conflict: deleting a row that is already deleted 10 -CASE MultiTransactionRelInsert --STATEMENT CALL ENABLE_MULTI_WRITES=true; +-STATEMENT CALL debug_enable_multi_writes=true; ---- ok -STATEMENT CALL auto_checkpoint=false; ---- ok @@ -275,7 +275,7 @@ Runtime exception: Write-write conflict: deleting a row that is already deleted -CASE MultiTransactionNodeCopy -SKIP --STATEMENT CALL ENABLE_MULTI_WRITES=true; +-STATEMENT CALL debug_enable_multi_writes=true; ---- ok -STATEMENT create node table person (ID INt64, fName StRING, gender INT64, isStudent BoOLEAN, isWorker BOOLEAN, age INT64, eyeSight DOUBLE, birthdate DATE, registerTime TIMESTAMP, lastJobDuration interval, workedHours INT64[], usedNames STRING[], courseScoresPerTerm INT64[][], grades INT64[4], height float, u UUID, PRIMARY KEY (ID)); ---- ok @@ -324,7 +324,7 @@ Runtime exception: Write-write conflict: deleting a row that is already deleted -CASE MultiTransactionRelCopy -SKIP --STATEMENT CALL ENABLE_MULTI_WRITES=true; +-STATEMENT CALL debug_enable_multi_writes=true; ---- ok -INSERT_STATEMENT_BLOCK COPY_TINYSNB_PERSON -STATEMENT create rel table knows (FROM person TO person, date DATE, meetTime TIMESTAMP, validInterval INTERVAL, comments STRING[], summary STRUCT(locations STRING[], transfer STRUCT(day DATE, amount INT64[])), notes UNION(firstmet DATE, type INT16, comment STRING), someMap MAP(STRING, STRING), MANY_MAnY); @@ -391,7 +391,7 @@ Runtime exception: Write-write conflict: deleting a row that is already deleted -CASE MixedMultiTransactionRelCopyInsert -SKIP --STATEMENT CALL ENABLE_MULTI_WRITES=true; +-STATEMENT CALL debug_enable_multi_writes=true; ---- ok -INSERT_STATEMENT_BLOCK COPY_TINYSNB_PERSON -STATEMENT create rel table knows (FROM person TO person, date DATE, meetTime TIMESTAMP, validInterval INTERVAL, comments STRING[], summary STRUCT(locations STRING[], transfer STRUCT(day DATE, amount INT64[])), notes UNION(firstmet DATE, type INT16, comment STRING), someMap MAP(STRING, STRING), MANY_MAnY);