Make storage version upgrades a ceremony (warn, opt-out flag, strict version mapping)

src/include/main/database.h

@@ -73,13 +73,17 @@ struct LBUG_API SystemConfig {
      * Default to false.
      * @param enableDefaultHashIndex If true, node tables create the default primary-key hash
      * index.
+     * @param allowStorageVersionUpgrade If true (default), checkpointing a database file written
+     * by an older Lbug release silently upgrades the file to the current storage version, after
+     * which older Lbug binaries can no longer open it. If false, such a checkpoint throws
+     * instead of upgrading, and the file stays readable by the release that wrote it.
      */
     explicit SystemConfig(uint64_t bufferPoolSize = -1u, uint64_t maxNumThreads = 0,
         bool enableCompression = true, bool readOnly = false, uint64_t maxDBSize = -1u,
         bool autoCheckpoint = true, uint64_t checkpointThreshold = 16777216 /* 16MB */,
         bool forceCheckpointOnClose = true, bool throwOnWalReplayFailure = false,
         bool enableChecksums = true, bool enableMultiWrites = false,
-        bool enableDefaultHashIndex = true
+        bool enableDefaultHashIndex = true, bool allowStorageVersionUpgrade = true
 #if defined(__APPLE__)
         ,
         uint32_t threadQos = QOS_CLASS_DEFAULT
@@ -98,6 +102,7 @@ struct LBUG_API SystemConfig {
     bool enableChecksums;
     bool enableMultiWrites;
     bool enableDefaultHashIndex;
+    bool allowStorageVersionUpgrade;
 #if defined(__APPLE__)
     uint32_t threadQos;
 #endif

src/include/main/db_config.h

@@ -29,6 +29,7 @@ struct DBConfig {
     bool enableChecksums;
     bool enableDefaultHashIndex;
     bool enableSpillingToDisk;
+    bool allowStorageVersionUpgrade;
 #if defined(__APPLE__)
     uint32_t threadQos;
 #endif

src/include/main/settings.h

@@ -124,6 +124,13 @@ struct ForceCheckpointClosingDBSetting {
     static common::Value getSetting(const ClientContext* context);
 };
 
+struct AllowStorageVersionUpgradeSetting {
+    static constexpr auto name = "allow_storage_version_upgrade";
+    static constexpr auto inputType = common::LogicalTypeID::BOOL;
+    static void setContext(ClientContext* context, const common::Value& parameter);
+    static common::Value getSetting(const ClientContext* context);
+};
+
 struct EnableDefaultHashIndexSetting {
     static constexpr auto name = "enable_default_hash_index";
     static constexpr auto inputType = common::LogicalTypeID::BOOL;

src/include/storage/storage_version_info.h

@@ -39,6 +39,8 @@ struct StorageVersionInfo {
     }
 
     static LBUG_API storage_version_t getStorageVersion();
+    static LBUG_API storage_version_t getStorageVersionForVersionString(
+        const std::string& rawVersion);
     static bool canReadStorageVersion(storage_version_t storageVersion) {
         return storageVersion == STORAGE_VERSION_40 || storageVersion == STORAGE_VERSION_41 ||
                storageVersion == STORAGE_VERSION_42 || storageVersion == STORAGE_VERSION_43 ||

src/main/database.cpp

@@ -35,7 +35,7 @@ namespace main {
 SystemConfig::SystemConfig(uint64_t bufferPoolSize_, uint64_t maxNumThreads, bool enableCompression,
     bool readOnly, uint64_t maxDBSize, bool autoCheckpoint, uint64_t checkpointThreshold,
     bool forceCheckpointOnClose, bool throwOnWalReplayFailure, bool enableChecksums,
-    bool enableMultiWrites, bool enableDefaultHashIndex
+    bool enableMultiWrites, bool enableDefaultHashIndex, bool allowStorageVersionUpgrade
 #if defined(__APPLE__)
     ,
     uint32_t threadQos
@@ -45,7 +45,8 @@ SystemConfig::SystemConfig(uint64_t bufferPoolSize_, uint64_t maxNumThreads, boo
       autoCheckpoint{autoCheckpoint}, checkpointThreshold{checkpointThreshold},
       forceCheckpointOnClose{forceCheckpointOnClose},
       throwOnWalReplayFailure(throwOnWalReplayFailure), enableChecksums(enableChecksums),
-      enableMultiWrites{enableMultiWrites}, enableDefaultHashIndex{enableDefaultHashIndex} {
+      enableMultiWrites{enableMultiWrites}, enableDefaultHashIndex{enableDefaultHashIndex},
+      allowStorageVersionUpgrade{allowStorageVersionUpgrade} {
 #if defined(__APPLE__)
     this->threadQos = threadQos;
 #endif

src/main/db_config.cpp

@@ -22,6 +22,7 @@ static ConfigurationOption options[] = { // NOLINT(cert-err58-cpp):
     GET_CONFIGURATION(RecursivePatternFactorSetting), GET_CONFIGURATION(EnableMVCCSetting),
     GET_CONFIGURATION(CheckpointThresholdSetting), GET_CONFIGURATION(AutoCheckpointSetting),
     GET_CONFIGURATION(ForceCheckpointClosingDBSetting),
+    GET_CONFIGURATION(AllowStorageVersionUpgradeSetting),
     GET_CONFIGURATION(EnableDefaultHashIndexSetting), GET_CONFIGURATION(SpillToDiskSetting),
     GET_CONFIGURATION(PKValidatorSpillThresholdSetting), GET_CONFIGURATION(EnableOptimizerSetting),
     GET_CONFIGURATION(EnableInternalCatalogSetting),
@@ -36,7 +37,8 @@ DBConfig::DBConfig(const SystemConfig& systemConfig)
       forceCheckpointOnClose{systemConfig.forceCheckpointOnClose},
       throwOnWalReplayFailure(systemConfig.throwOnWalReplayFailure),
       enableChecksums(systemConfig.enableChecksums),
-      enableDefaultHashIndex{systemConfig.enableDefaultHashIndex}, enableSpillingToDisk{true} {
+      enableDefaultHashIndex{systemConfig.enableDefaultHashIndex}, enableSpillingToDisk{true},
+      allowStorageVersionUpgrade{systemConfig.allowStorageVersionUpgrade} {
 #if defined(__APPLE__)
     this->threadQos = systemConfig.threadQos;
 #endif

src/main/settings.cpp

@@ -200,6 +200,16 @@ common::Value ForceCheckpointClosingDBSetting::getSetting(const ClientContext* c
     return common::Value(context->getDBConfig()->forceCheckpointOnClose);
 }
 
+void AllowStorageVersionUpgradeSetting::setContext(ClientContext* context,
+    const common::Value& parameter) {
+    parameter.validateType(inputType);
+    context->getDBConfigUnsafe()->allowStorageVersionUpgrade = parameter.getValue<bool>();
+}
+
+common::Value AllowStorageVersionUpgradeSetting::getSetting(const ClientContext* context) {
+    return common::Value(context->getDBConfig()->allowStorageVersionUpgrade);
+}
+
 void EnableDefaultHashIndexSetting::setContext(ClientContext* context,
     const common::Value& parameter) {
     parameter.validateType(inputType);

src/storage/checkpointer.cpp

@@ -1,6 +1,7 @@
 #include "storage/checkpointer.h"
 
 #include <chrono>
+#include <iostream>
 #include <string_view>
 #include <thread>
 #include <vector>
@@ -228,6 +229,25 @@ PageRange Checkpointer::serializeMetadata(const catalog::Catalog& catalog,
     return allocatedPages;
 }
 
+static void applyStorageVersionUpgradeCeremony(const main::ClientContext& clientContext,
+    storage_version_t oldStorageVersion, storage_version_t newStorageVersion) {
+    if (oldStorageVersion == newStorageVersion) {
+        return;
+    }
+    if (!clientContext.getDBConfig()->allowStorageVersionUpgrade) {
+        throw common::RuntimeException(std::format(
+            "Checkpoint would upgrade the database storage version from {} to {}, after which "
+            "older Lbug binaries can no longer open this database file, and "
+            "allow_storage_version_upgrade is false. Run CALL "
+            "allow_storage_version_upgrade=true; to allow the upgrade.",
+            oldStorageVersion, newStorageVersion));
+    }
+    std::cerr << "Warning: upgrading database storage version from " << oldStorageVersion << " to "
+              << newStorageVersion
+              << ". Older Lbug binaries will no longer be able to open this database file."
+              << std::endl;
+}
+
 void Checkpointer::writeCheckpoint() {
     if (isInMemory) {
         return;
@@ -236,16 +256,21 @@ void Checkpointer::writeCheckpoint() {
     acquireCheckpointLocks();
     checkpointTargets = collectCheckpointTargets();
 
+    auto databaseHeader = *mainStorageManager->getOrInitDatabaseHeader(clientContext);
+    const auto oldStorageVersion = databaseHeader.storageVersion;
+    databaseHeader.storageVersion = StorageVersionInfo::getStorageVersion();
+    hasStorageVersionUpgrade = oldStorageVersion != databaseHeader.storageVersion;
+    // Refuse (or warn about) a storage-version upgrade before any WAL rotation: an exception
+    // thrown after rotation would leave a frozen WAL that replays over a later checkpoint.
+    applyStorageVersionUpgradeCeremony(clientContext, oldStorageVersion,
+        databaseHeader.storageVersion);
+
     for (const auto& target : checkpointTargets) {
         auto rotated = target.storageManager->getWAL().rotateForCheckpoint(&clientContext);
         walRotatedByManager[target.storageManager] = rotated;
         walRotated = walRotated || rotated;
     }
 
-    auto databaseHeader = *mainStorageManager->getOrInitDatabaseHeader(clientContext);
-    const auto oldStorageVersion = databaseHeader.storageVersion;
-    databaseHeader.storageVersion = StorageVersionInfo::getStorageVersion();
-    hasStorageVersionUpgrade = oldStorageVersion != databaseHeader.storageVersion;
     bool localHasStorageChanges = checkpointStorage();
     serializeCatalogAndMetadata(databaseHeader, localHasStorageChanges);
     databaseHeader.dataFileNumPages = mainStorageManager->getDataFH()->getNumPages();
@@ -281,17 +306,21 @@ void Checkpointer::beginCheckpoint(common::transaction_t snapshotTimestamp) {
     snapshotTS = snapshotTimestamp;
     checkpointTargets = collectCheckpointTargets();
 
+    checkpointHeader = *mainStorageManager->getOrInitDatabaseHeader(clientContext);
+    const auto oldStorageVersion = checkpointHeader.storageVersion;
+    checkpointHeader.storageVersion = StorageVersionInfo::getStorageVersion();
+    hasStorageVersionUpgrade = oldStorageVersion != checkpointHeader.storageVersion;
+    // Refuse (or warn about) a storage-version upgrade before any WAL rotation: an exception
+    // thrown after rotation would leave a frozen WAL that replays over a later checkpoint.
+    applyStorageVersionUpgradeCeremony(clientContext, oldStorageVersion,
+        checkpointHeader.storageVersion);
+
     for (const auto& target : checkpointTargets) {
         auto rotated = target.storageManager->getWAL().rotateForCheckpoint(&clientContext);
         walRotatedByManager[target.storageManager] = rotated;
         walRotated = walRotated || rotated;
     }
 
-    checkpointHeader = *mainStorageManager->getOrInitDatabaseHeader(clientContext);
-    const auto oldStorageVersion = checkpointHeader.storageVersion;
-    checkpointHeader.storageVersion = StorageVersionInfo::getStorageVersion();
-    hasStorageVersionUpgrade = oldStorageVersion != checkpointHeader.storageVersion;
-
     // Capture versions while the write gate is still held.
     for (const auto& target : checkpointTargets) {
         catalogVersionAtCheckpointByCatalog[target.catalog] = target.catalog->getVersion();

src/storage/database_header.cpp

@@ -21,7 +21,6 @@ static storage_version_t validateStorageVersion(common::Deserializer& deSer) {
     deSer.deserializeValue(savedStorageVersion);
     const auto storageVersion = StorageVersionInfo::getStorageVersion();
     if (!StorageVersionInfo::canReadStorageVersion(savedStorageVersion)) {
-        // TODO(Guodong): Add a test case for this.
         throw common::RuntimeException(
             std::format("Trying to read a database file with a different version. "
                         "Database file version: {}, Current build storage version: {}",
@@ -85,8 +84,7 @@ void DatabaseHeader::serialize(common::Serializer& ser) const {
     ser.serializeValue(dataFileNumPages);
 }
 
-DatabaseHeader DatabaseHeader::deserialize(common::Deserializer& deSer) {
-    validateMagicBytes(deSer);
+static DatabaseHeader deserializeHeaderBody(common::Deserializer& deSer) {
     const auto savedStorageVersion = validateStorageVersion(deSer);
     PageRange catalogPageRange{}, metaPageRange{};
     common::uuid databaseID{};
@@ -110,6 +108,11 @@ DatabaseHeader DatabaseHeader::deserialize(common::Deserializer& deSer) {
     return {catalogPageRange, metaPageRange, dataFileNumPages, databaseID, savedStorageVersion};
 }
 
+DatabaseHeader DatabaseHeader::deserialize(common::Deserializer& deSer) {
+    validateMagicBytes(deSer);
+    return deserializeHeaderBody(deSer);
+}
+
 DatabaseHeader DatabaseHeader::createInitialHeader(common::RandomEngine* randomEngine) {
     // We generate a random UUID to act as the database ID
     return DatabaseHeader{{}, {}, 0, common::UUID::generateRandomUUID(randomEngine),
@@ -124,11 +127,15 @@ std::optional<DatabaseHeader> DatabaseHeader::readDatabaseHeader(common::FileInf
     auto reader = std::make_unique<common::BufferedFileReader>(dataFileInfo);
     common::Deserializer deSer(std::move(reader));
     try {
-        return DatabaseHeader::deserialize(deSer);
+        validateMagicBytes(deSer);
     } catch (const common::RuntimeException&) {
         // It is possible we optimistically write to the database file before the first checkpoint
         // In this case the magic bytes check will fail and we assume there is no existing header
         return std::nullopt;
     }
+    // Past the magic bytes the file is a Lbug database file: any further failure (e.g. an
+    // unreadable storage version) is a real error and must propagate rather than be treated as
+    // "no existing header".
+    return deserializeHeaderBody(deSer);
 }
 } // namespace lbug::storage

src/storage/storage_version_info.cpp

@@ -1,5 +1,8 @@
 #include "storage/storage_version_info.h"
 
+#include "common/exception/runtime.h"
+#include <format>
+
 namespace lbug {
 namespace storage {
 
@@ -19,25 +22,26 @@ static bool usesStorageVersion40(const std::string& version) {
            version.starts_with("0.16.");
 }
 
-storage_version_t StorageVersionInfo::getStorageVersion() {
+storage_version_t StorageVersionInfo::getStorageVersionForVersionString(
+    const std::string& rawVersion) {
     auto storageVersionInfo = getStorageVersionInfo();
-    auto version = normalizeVersionForStorageLookup(LBUG_CMAKE_VERSION);
+    auto version = normalizeVersionForStorageLookup(rawVersion);
     if (usesStorageVersion40(version)) {
         return STORAGE_VERSION_40;
     }
     if (!storageVersionInfo.contains(version)) {
-        // If the current LBUG_CMAKE_VERSION is not in the map,
-        // then we must run the newest version of lbug
-        // LCOV_EXCL_START
-        storage_version_t maxVersion = 0;
-        for (auto& [_, versionNumber] : storageVersionInfo) {
-            maxVersion = std::max(maxVersion, versionNumber);
-        }
-        return maxVersion;
-        // LCOV_EXCL_STOP
+        throw common::RuntimeException(
+            std::format("Lbug version '{}' has no storage version mapping. This is a build "
+                        "configuration error: add the release to "
+                        "StorageVersionInfo::getStorageVersionInfo().",
+                rawVersion));
     }
     return storageVersionInfo.at(version);
 }
 
+storage_version_t StorageVersionInfo::getStorageVersion() {
+    return getStorageVersionForVersionString(LBUG_CMAKE_VERSION);
+}
+
 } // namespace storage
 } // namespace lbug

test/api/CMakeLists.txt

@@ -15,4 +15,5 @@ add_lbug_api_test(api_test
         result_value_test.cpp
         storage_driver_test.cpp
         udf_test.cpp
-        read_only_test.cpp)
+        read_only_test.cpp
+        storage_version_ceremony_test.cpp)