PageStorage: Add mix mode

dbms/src/Debug/dbgFuncMisc.cpp

@@ -104,9 +104,7 @@ void dbgFuncTriggerGlobalPageStorageGC(Context & context, const ASTs & /*args*/,
     auto global_storage_pool = context.getGlobalStoragePool();
     if (global_storage_pool)
     {
-        global_storage_pool->meta()->gc();
-        global_storage_pool->log()->gc();
-        global_storage_pool->data()->gc();
+        global_storage_pool->gc();
     }
 }
 } // namespace DB

dbms/src/Interpreters/Context.cpp

@@ -161,6 +161,7 @@ struct ContextShared
     PathCapacityMetricsPtr path_capacity_ptr; /// Path capacity metrics
     FileProviderPtr file_provider; /// File provider.
     IORateLimiter io_rate_limiter;
+    PageStorageRunMode storage_run_mode;
     DM::GlobalStoragePoolPtr global_storage_pool;
     /// Named sessions. The user could specify session identifier to reuse settings and temporary tables in subsequent requests.
 
@@ -1547,24 +1548,38 @@ ReadLimiterPtr Context::getReadLimiter() const
     return getIORateLimiter().getReadLimiter();
 }
 
-static bool isUsingPageStorageV3(const PathPool & path_pool, bool enable_ps_v3)
+
+static bool isPageStorageV2Existed(const PathPool & path_pool)
 {
-    // Check whether v3 is already enabled
-    for (const auto & path : path_pool.listGlobalPagePaths())
+    for (const auto & path : path_pool.listKVStorePaths())
     {
-        if (PS::V3::PageStorageImpl::isManifestsFileExists(path))
+        Poco::File dir(path);
+        if (!dir.exists())
+            continue;
+
+        std::vector<std::string> files;
+        dir.list(files);
+        if (!files.empty())
         {
-            return true;
+            for (const auto & file_name : files)
+            {
+                const auto & find_index = file_name.find("page");
+                if (find_index != std::string::npos)
+                {
+                    return true;
+                }
+            }
+            // KVStore is not empty, but can't find any of v2 data in it.
         }
     }
 
-    // Check whether v3 on new node is enabled in the config, if not, no need to check anymore
-    if (!enable_ps_v3)
-        return false;
+    // If not data in KVStore. It means V2 data must not existed.
+    return false;
+}
 
-    // Check whether there are any files in kvstore path, if exists, then this is not a new node.
-    // If it's a new node, then we enable v3. Otherwise, we use v2.
-    for (const auto & path : path_pool.listKVStorePaths())
+static bool isPageStorageV3Existed(const PathPool & path_pool)
+{
+    for (const auto & path : path_pool.listGlobalPagePaths())
     {
         Poco::File dir(path);
         if (!dir.exists())
@@ -1574,23 +1589,76 @@ static bool isUsingPageStorageV3(const PathPool & path_pool, bool enable_ps_v3)
         dir.list(files);
         if (!files.empty())
         {
-            return false;
+            return true;
         }
     }
-    return true;
+    return false;
 }
 
-bool Context::initializeGlobalStoragePoolIfNeed(const PathPool & path_pool, bool enable_ps_v3)
+void Context::initializePageStorageMode(const PathPool & path_pool, UInt64 storage_page_format_version)
 {
     auto lock = getLock();
-    if (isUsingPageStorageV3(path_pool, enable_ps_v3))
+
+    /**
+     * PageFormat::V2 + isPageStorageV3Existed is false + whatever isPageStorageV2Existed true or false = ONLY_V2
+     * PageFormat::V2 + isPageStorageV3Existed is true  + whatever isPageStorageV2Existed true or false = ERROR Config
+     * PageFormat::V3 + isPageStorageV2Existed is true  + whatever isPageStorageV3Existed true or false = MIX_MODE
+     * PageFormat::V3 + isPageStorageV2Existed is false + whatever isPageStorageV3Existed true or false = ONLY_V3
+     */
+
+    switch (storage_page_format_version)
     {
-        try
+    case PageFormat::V1:
+    case PageFormat::V2:
+    {
+        if (isPageStorageV3Existed(path_pool))
         {
-            // create manifests file before initialize GlobalStoragePool
-            for (const auto & path : path_pool.listGlobalPagePaths())
-                PS::V3::PageStorageImpl::createManifestsFileIfNeed(path);
+            throw Exception("Invalid config `storage.format_version`, Current page V3 data exist. But using the PageFormat::V2."
+                            "If you are downgrading the format_version for this TiFlash node, you need to rebuild the data from scratch.",
+                            ErrorCodes::LOGICAL_ERROR);
+        }
+        // not exist V3
+        shared->storage_run_mode = PageStorageRunMode::ONLY_V2;
+        return;
+    }
+    case PageFormat::V3:
+    {
+        shared->storage_run_mode = isPageStorageV2Existed(path_pool) ? PageStorageRunMode::MIX_MODE : PageStorageRunMode::ONLY_V3;
+        return;
+    }
+    default:
+        throw Exception(fmt::format("Can't detect the format version of Page [page_version={}]", storage_page_format_version),
+                        ErrorCodes::LOGICAL_ERROR);
+    }
+}
+
+PageStorageRunMode Context::getPageStorageRunMode() const
+{
+    auto lock = getLock();
+    return shared->storage_run_mode;
+}
 
+void Context::setPageStorageRunMode(PageStorageRunMode run_mode) const
+{
+    auto lock = getLock();
+    shared->storage_run_mode = run_mode;
+}
+
+bool Context::initializeGlobalStoragePoolIfNeed(const PathPool & path_pool)
+{
+    auto lock = getLock();
+    if (shared->global_storage_pool)
+    {
+        // Can't init GlobalStoragePool twice.
+        // Because we won't remove the gc task in BackGroundPool
+        // Also won't remove it from ~GlobalStoragePool()
+        throw Exception("GlobalStoragePool has already been initialized.", ErrorCodes::LOGICAL_ERROR);
+    }
+
+    if (shared->storage_run_mode == PageStorageRunMode::MIX_MODE || shared->storage_run_mode == PageStorageRunMode::ONLY_V3)
+    {
+        try
+        {
             shared->global_storage_pool = std::make_shared<DM::GlobalStoragePool>(path_pool, *this, settings);
             shared->global_storage_pool->restore();
             return true;

dbms/src/Interpreters/Context.h

@@ -99,6 +99,7 @@ using WriteLimiterPtr = std::shared_ptr<WriteLimiter>;
 class ReadLimiter;
 using ReadLimiterPtr = std::shared_ptr<ReadLimiter>;
 
+enum class PageStorageRunMode : UInt8;
 namespace DM
 {
 class MinMaxIndexCache;
@@ -405,8 +406,10 @@ class Context
     ReadLimiterPtr getReadLimiter() const;
     IORateLimiter & getIORateLimiter() const;
 
-    bool initializeGlobalStoragePoolIfNeed(const PathPool & path_pool, bool enable_ps_v3);
-
+    void initializePageStorageMode(const PathPool & path_pool, UInt64 storage_page_format_version);
+    void setPageStorageRunMode(PageStorageRunMode run_mode) const;
+    PageStorageRunMode getPageStorageRunMode() const;
+    bool initializeGlobalStoragePoolIfNeed(const PathPool & path_pool);
     DM::GlobalStoragePoolPtr getGlobalStoragePool() const;
 
     /// Call after initialization before using system logs. Call for global context.

dbms/src/Server/DTTool/DTToolBench.cpp

@@ -336,7 +336,7 @@ int benchEntry(const std::vector<std::string> & opts)
         auto settings = DB::Settings();
         auto db_context = env.getContext();
         auto path_pool = std::make_unique<DB::StoragePathPool>(db_context->getPathPool().withTable("test", "t1", false));
-        auto storage_pool = std::make_unique<DB::DM::StoragePool>("test.t1", /*table_id*/ 1, *path_pool, *db_context, db_context->getSettingsRef());
+        auto storage_pool = std::make_unique<DB::DM::StoragePool>(*db_context, /*ns_id*/ 1, *path_pool, "test.t1");
         auto dm_settings = DB::DM::DeltaMergeStore::Settings{};
         auto dm_context = std::make_unique<DB::DM::DMContext>( //
             *db_context,

dbms/src/Server/Server.cpp

@@ -1112,13 +1112,10 @@ int Server::main(const std::vector<std::string> & /*args*/)
         raft_config.enable_compatible_mode, //
         global_context->getPathCapacity(),
         global_context->getFileProvider());
-    // must initialize before the following operation:
-    //   1. load data from disk(because this process may depend on the initialization of global StoragePool)
-    //   2. initialize KVStore service
-    //     1) because we need to check whether this is the first startup of this node, and we judge it based on whether there are any files in kvstore directory
-    //     2) KVStore service also choose its data format based on whether the GlobalStoragePool is initialized
-    if (global_context->initializeGlobalStoragePoolIfNeed(global_context->getPathPool(), storage_config.enable_ps_v3))
-        LOG_FMT_INFO(log, "PageStorage V3 enabled.");
+
+    global_context->initializePageStorageMode(global_context->getPathPool(), STORAGE_FORMAT_CURRENT.page);
+    global_context->initializeGlobalStoragePoolIfNeed(global_context->getPathPool());
+    LOG_FMT_INFO(log, "Global PageStorage run mode is {}", static_cast<UInt8>(global_context->getPageStorageRunMode()));
 
     // Use pd address to define which default_database we use by default.
     // For mock test, we use "default". For deployed with pd/tidb/tikv use "system", which is always exist in TiFlash.

dbms/src/Server/StorageConfigParser.cpp

@@ -213,10 +213,8 @@ void TiFlashStorageConfig::parseMisc(const String & storage_section, Poco::Logge
     };
 
     lazily_init_store = get_bool_config_or_default("lazily_init_store", lazily_init_store);
-    // config for experimental feature, may remove later
-    enable_ps_v3 = get_bool_config_or_default("enable_ps_v3", enable_ps_v3);
 
-    LOG_FMT_INFO(log, "format_version {} lazily_init_store {} enable_ps_v3 {}", format_version, lazily_init_store, enable_ps_v3);
+    LOG_FMT_INFO(log, "format_version {} lazily_init_store {}", format_version, lazily_init_store);
 }
 
 Strings TiFlashStorageConfig::getAllNormalPaths() const

dbms/src/Server/StorageConfigParser.h

@@ -103,7 +103,6 @@ struct TiFlashStorageConfig
 
     UInt64 format_version = 0;
     bool lazily_init_store = true;
-    bool enable_ps_v3 = true;
 
 public:
     TiFlashStorageConfig() = default;

dbms/src/Server/tests/gtest_dttool.cpp

@@ -72,7 +72,7 @@ struct DTToolTest : public DB::base::TiFlashStorageTestBasic
             properties.push_back(property);
         }
         auto path_pool = std::make_unique<DB::StoragePathPool>(db_context->getPathPool().withTable("test", "t1", false));
-        auto storage_pool = std::make_unique<DB::DM::StoragePool>("test.t1", /*table_id*/ 1, *path_pool, *db_context, db_context->getSettingsRef());
+        auto storage_pool = std::make_unique<DB::DM::StoragePool>(*db_context, /*ns_id*/ 1, *path_pool, "test.t1");
         auto dm_settings = DB::DM::DeltaMergeStore::Settings{};
         auto dm_context = std::make_unique<DB::DM::DMContext>( //
             *db_context,

dbms/src/Server/tests/gtest_server_config.cpp

@@ -296,7 +296,7 @@ dt_page_gc_low_write_prob = 0.2
     auto verify_persister_reload_config = [&global_ctx](RegionPersister & persister) {
         DB::Settings & settings = global_ctx.getSettingsRef();
 
-        auto cfg = persister.page_storage->getSettings();
+        auto cfg = persister.getPageStorageSettings();
         EXPECT_NE(cfg.gc_min_files, settings.dt_storage_pool_data_gc_min_file_num);
         EXPECT_NE(cfg.gc_min_legacy_num, settings.dt_storage_pool_data_gc_min_legacy_num);
         EXPECT_NE(cfg.gc_min_bytes, settings.dt_storage_pool_data_gc_min_bytes);
@@ -307,8 +307,7 @@ dt_page_gc_low_write_prob = 0.2
         EXPECT_NE(cfg.prob_do_gc_when_write_is_low, settings.dt_page_gc_low_write_prob * 1000);
         persister.gc();
 
-        cfg = persister.page_storage->getSettings();
-
+        cfg = persister.getPageStorageSettings();
         EXPECT_NE(cfg.gc_min_files, settings.dt_storage_pool_data_gc_min_file_num);
         EXPECT_NE(cfg.gc_min_legacy_num, settings.dt_storage_pool_data_gc_min_legacy_num);
         EXPECT_NE(cfg.gc_min_bytes, settings.dt_storage_pool_data_gc_min_bytes);
@@ -370,12 +369,12 @@ dt_page_gc_low_write_prob = 0.2
 
     auto & global_ctx = TiFlashTestEnv::getGlobalContext();
     std::unique_ptr<StoragePathPool> path_pool = std::make_unique<StoragePathPool>(global_ctx.getPathPool().withTable("test", "t1", false));
-    std::unique_ptr<DM::StoragePool> storage_pool = std::make_unique<DM::StoragePool>("test.t1", /*table_id*/ 100, *path_pool, global_ctx, global_ctx.getSettingsRef());
+    std::unique_ptr<DM::StoragePool> storage_pool = std::make_unique<DM::StoragePool>(global_ctx, /*ns_id*/ 100, *path_pool, "test.t1");
 
     auto verify_storage_pool_reload_config = [&global_ctx](std::unique_ptr<DM::StoragePool> & storage_pool) {
         DB::Settings & settings = global_ctx.getSettingsRef();
 
-        auto cfg = storage_pool->data()->getSettings();
+        auto cfg = storage_pool->data_storage_v2->getSettings();
         EXPECT_NE(cfg.gc_min_files, settings.dt_storage_pool_data_gc_min_file_num);
         EXPECT_NE(cfg.gc_min_legacy_num, settings.dt_storage_pool_data_gc_min_legacy_num);
         EXPECT_NE(cfg.gc_min_bytes, settings.dt_storage_pool_data_gc_min_bytes);
@@ -387,7 +386,7 @@ dt_page_gc_low_write_prob = 0.2
 
         storage_pool->gc(settings, DM::StoragePool::Seconds(0));
 
-        cfg = storage_pool->data()->getSettings();
+        cfg = storage_pool->data_storage_v2->getSettings();
         EXPECT_EQ(cfg.gc_min_files, settings.dt_storage_pool_data_gc_min_file_num);
         EXPECT_EQ(cfg.gc_min_legacy_num, settings.dt_storage_pool_data_gc_min_legacy_num);
         EXPECT_EQ(cfg.gc_min_bytes, settings.dt_storage_pool_data_gc_min_bytes);

dbms/src/Storages/DeltaMerge/ColumnFile/ColumnFileBig.cpp

@@ -72,7 +72,7 @@ ColumnFilePersistedPtr ColumnFileBig::deserializeMetadata(DMContext & context, /
     readIntBinary(valid_rows, buf);
     readIntBinary(valid_bytes, buf);
 
-    auto file_id = context.storage_pool.dataReader().getNormalPageId(file_ref_id);
+    auto file_id = context.storage_pool.dataReader()->getNormalPageId(file_ref_id);
     auto file_parent_path = context.path_pool.getStableDiskDelegator().getDTFilePath(file_id);
 
     auto dmfile = DMFile::restore(context.db_context.getFileProvider(), file_id, file_ref_id, file_parent_path, DMFile::ReadMetaMode::all());

dbms/src/Storages/DeltaMerge/Delta/ColumnFilePersistedSet.cpp

@@ -107,7 +107,7 @@ ColumnFilePersistedSet::ColumnFilePersistedSet(PageId metadata_id_, const Column
 
 ColumnFilePersistedSetPtr ColumnFilePersistedSet::restore(DMContext & context, const RowKeyRange & segment_range, PageId id)
 {
-    Page page = context.storage_pool.metaReader().read(id);
+    Page page = context.storage_pool.metaReader()->read(id);
     ReadBufferFromMemory buf(page.data.begin(), page.data.size());
     auto column_files = deserializeSavedColumnFiles(context, segment_range, buf);
     return std::make_shared<ColumnFilePersistedSet>(id, column_files);

dbms/src/Storages/DeltaMerge/Delta/DeltaValueSpace.cpp

@@ -236,12 +236,12 @@ bool DeltaValueSpace::compact(DMContext & context)
             LOG_FMT_DEBUG(log, "{} Nothing to compact", simpleInfo());
             return true;
         }
-        log_storage_snap = context.storage_pool.log()->getSnapshot(/*tracing_id*/ fmt::format("minor_compact_{}", simpleInfo()));
+        log_storage_snap = context.storage_pool.logReader()->getSnapshot(/*tracing_id*/ fmt::format("minor_compact_{}", simpleInfo()));
     }
 
     // do compaction task
     WriteBatches wbs(context.storage_pool, context.getWriteLimiter());
-    PageReader reader(context.storage_pool.getNamespaceId(), context.storage_pool.log(), std::move(log_storage_snap), context.getReadLimiter());
+    const auto & reader = context.storage_pool.newLogReader(context.getReadLimiter(), log_storage_snap);
     compaction_task->prepare(context, wbs, reader);
 
     {