Refactor: add new module schema

dbms/CMakeLists.txt

@@ -90,6 +90,7 @@ add_headers_and_sources(dbms src/Storages/Page/V2/mvcc)
 add_headers_and_sources(dbms src/Storages/Page/V2/VersionSet)
 add_headers_and_sources(dbms src/Storages/Page/V2/gc)
 add_headers_and_sources(dbms src/WindowFunctions)
+add_headers_and_sources(dbms src/TiDB/Schema)
 if (ENABLE_V3_PAGESTORAGE)
     add_headers_and_sources(dbms src/Storages/Page/V3)
     add_headers_and_sources(dbms src/Storages/Page/V3/LogFile)

dbms/src/Databases/test/gtest_database.cpp

@@ -25,11 +25,11 @@
 #include <Storages/IManageableStorage.h>
 #include <Storages/IStorage.h>
 #include <Storages/MutableSupport.h>
-#include <Storages/Transaction/SchemaNameMapper.h>
 #include <Storages/Transaction/TMTContext.h>
 #include <Storages/Transaction/TMTStorages.h>
 #include <Storages/registerStorages.h>
 #include <TestUtils/TiFlashTestBasic.h>
+#include <TiDB/Schema/SchemaNameMapper.h>
 #include <common/ThreadPool.h>
 #include <common/logger_useful.h>
 

dbms/src/Debug/MockSchemaGetter.h

@@ -15,14 +15,13 @@
 #pragma once
 
 #include <Debug/MockTiDB.h>
-#include <Storages/Transaction/SchemaGetter.h>
+#include <TiDB/Schema/SchemaGetter.h>
 
 namespace DB
 {
 
 struct MockSchemaGetter
 {
-
     TiDB::DBInfoPtr getDatabase(DatabaseID db_id) { return MockTiDB::instance().getDBInfoByID(db_id); }
 
     Int64 getVersion() { return MockTiDB::instance().getVersion(); }

dbms/src/Debug/MockSchemaNameMapper.h

@@ -14,7 +14,7 @@
 
 #pragma once
 
-#include <Storages/Transaction/SchemaNameMapper.h>
+#include <TiDB/Schema/SchemaNameMapper.h>
 
 namespace DB
 {

dbms/src/Debug/MockTiDB.h

@@ -15,10 +15,10 @@
 #pragma once
 
 #include <Storages/ColumnsDescription.h>
-#include <Storages/Transaction/SchemaGetter.h>
-#include <Storages/Transaction/SchemaSyncer.h>
 #include <Storages/Transaction/TiDB.h>
 #include <Storages/Transaction/Types.h>
+#include <TiDB/Schema/SchemaGetter.h>
+#include <TiDB/Schema/SchemaSyncer.h>
 
 #include <atomic>
 

dbms/src/Debug/dbgFuncCoprocessor.cpp

@@ -336,7 +336,7 @@ void dbgFuncTiDBQueryFromNaturalDag(Context & context, const ASTs & args, DBGInv
     if (args.size() != 1)
         throw Exception("Args not matched, should be: json_dag_path", ErrorCodes::BAD_ARGUMENTS);
 
-    String json_dag_path = safeGet<String>(typeid_cast<const ASTLiteral &>(*args[0]).value);
+    auto json_dag_path = safeGet<String>(typeid_cast<const ASTLiteral &>(*args[0]).value);
     auto dag = NaturalDag(json_dag_path, &Poco::Logger::get("MockDAG"));
     dag.init();
     dag.build(context);
@@ -431,7 +431,7 @@ BlockInputStreamPtr dbgFuncTiDBQuery(Context & context, const ASTs & args)
     if (args.empty() || args.size() > 3)
         throw Exception("Args not matched, should be: query[, region-id, dag_prop_string]", ErrorCodes::BAD_ARGUMENTS);
 
-    String query = safeGet<String>(typeid_cast<const ASTLiteral &>(*args[0]).value);
+    auto query = safeGet<String>(typeid_cast<const ASTLiteral &>(*args[0]).value);
     RegionID region_id = InvalidRegionID;
     if (args.size() >= 2)
         region_id = safeGet<RegionID>(typeid_cast<const ASTLiteral &>(*args[1]).value);
@@ -464,8 +464,8 @@ BlockInputStreamPtr dbgFuncMockTiDBQuery(Context & context, const ASTs & args)
     if (args.size() < 2 || args.size() > 4)
         throw Exception("Args not matched, should be: query, region-id[, start-ts, dag_prop_string]", ErrorCodes::BAD_ARGUMENTS);
 
-    String query = safeGet<String>(typeid_cast<const ASTLiteral &>(*args[0]).value);
-    RegionID region_id = safeGet<RegionID>(typeid_cast<const ASTLiteral &>(*args[1]).value);
+    auto query = safeGet<String>(typeid_cast<const ASTLiteral &>(*args[0]).value);
+    auto region_id = safeGet<RegionID>(typeid_cast<const ASTLiteral &>(*args[1]).value);
     Timestamp start_ts = DEFAULT_MAX_READ_TSO;
     if (args.size() >= 3)
         start_ts = safeGet<Timestamp>(typeid_cast<const ASTLiteral &>(*args[2]).value);
@@ -671,14 +671,14 @@ const ASTTablesInSelectQueryElement * getJoin(ASTSelectQuery & ast_query)
     if (!ast_query.tables)
         return nullptr;
 
-    const ASTTablesInSelectQuery & tables_in_select_query = static_cast<const ASTTablesInSelectQuery &>(*ast_query.tables);
+    const auto & tables_in_select_query = static_cast<const ASTTablesInSelectQuery &>(*ast_query.tables);
     if (tables_in_select_query.children.empty())
         return nullptr;
 
     const ASTTablesInSelectQueryElement * joined_table = nullptr;
     for (const auto & child : tables_in_select_query.children)
     {
-        const ASTTablesInSelectQueryElement & tables_element = static_cast<const ASTTablesInSelectQueryElement &>(*child);
+        const auto & tables_element = static_cast<const ASTTablesInSelectQueryElement &>(*child);
         if (tables_element.table_join)
         {
             if (!joined_table)
@@ -737,7 +737,7 @@ std::pair<ExecutorPtr, bool> compileQueryBlock(
             bool append_pk_column = false;
             for (const auto & expr : ast_query.select_expression_list->children)
             {
-                if (ASTIdentifier * identifier = typeid_cast<ASTIdentifier *>(expr.get()))
+                if (auto * identifier = typeid_cast<ASTIdentifier *>(expr.get()))
                 {
                     if (identifier->getColumnName() == MutableSupport::tidb_pk_column_name)
                     {
@@ -756,7 +756,7 @@ std::pair<ExecutorPtr, bool> compileQueryBlock(
         String right_table_alias;
         {
             String database_name, table_name;
-            const ASTTableExpression & table_to_join = static_cast<const ASTTableExpression &>(*joined_table->table_expression);
+            const auto & table_to_join = static_cast<const ASTTableExpression &>(*joined_table->table_expression);
             if (table_to_join.database_and_table_name)
             {
                 auto identifier = static_cast<const ASTIdentifier &>(*table_to_join.database_and_table_name);
@@ -788,7 +788,7 @@ std::pair<ExecutorPtr, bool> compileQueryBlock(
         bool right_append_pk_column = false;
         for (const auto & expr : ast_query.select_expression_list->children)
         {
-            if (ASTIdentifier * identifier = typeid_cast<ASTIdentifier *>(expr.get()))
+            if (auto * identifier = typeid_cast<ASTIdentifier *>(expr.get()))
             {
                 auto names = splitQualifiedName(identifier->getColumnName());
                 if (names.second == MutableSupport::tidb_pk_column_name)
@@ -831,7 +831,7 @@ std::pair<ExecutorPtr, bool> compileQueryBlock(
     bool has_agg_func = false;
     for (const auto & child : ast_query.select_expression_list->children)
     {
-        const ASTFunction * func = typeid_cast<const ASTFunction *>(child.get());
+        const auto * func = typeid_cast<const ASTFunction *>(child.get());
         if (func && AggregateFunctionFactory::instance().isAggregateFunctionName(func->name))
         {
             has_agg_func = true;

dbms/src/Debug/dbgFuncMockTiDBTable.cpp

@@ -24,8 +24,8 @@
 #include <Parsers/ParserRenameQuery.h>
 #include <Parsers/parseQuery.h>
 #include <Storages/Transaction/KVStore.h>
-#include <Storages/Transaction/SchemaSyncer.h>
 #include <Storages/Transaction/TMTContext.h>
+#include <TiDB/Schema/SchemaSyncer.h>
 #include <fmt/core.h>
 
 namespace DB

dbms/src/Debug/dbgFuncSchema.cpp

@@ -22,10 +22,10 @@
 #include <Parsers/ASTLiteral.h>
 #include <Parsers/ParserCreateQuery.h>
 #include <Storages/IManageableStorage.h>
-#include <Storages/Transaction/SchemaSyncService.h>
-#include <Storages/Transaction/SchemaSyncer.h>
 #include <Storages/Transaction/TMTContext.h>
 #include <Storages/Transaction/TiDB.h>
+#include <TiDB/Schema/SchemaSyncService.h>
+#include <TiDB/Schema/SchemaSyncer.h>
 #include <fmt/core.h>
 
 #include <ext/singleton.h>

dbms/src/Debug/dbgFuncSchemaName.cpp

@@ -20,9 +20,9 @@
 #include <Parsers/ASTIdentifier.h>
 #include <Parsers/ASTLiteral.h>
 #include <Storages/IManageableStorage.h>
-#include <Storages/Transaction/SchemaNameMapper.h>
-#include <Storages/Transaction/SchemaSyncer.h>
 #include <Storages/Transaction/TMTContext.h>
+#include <TiDB/Schema/SchemaNameMapper.h>
+#include <TiDB/Schema/SchemaSyncer.h>
 #include <fmt/core.h>
 
 #include <boost/algorithm/string/replace.hpp>
@@ -97,7 +97,7 @@ BlockInputStreamPtr dbgFuncQueryMapped(Context & context, const ASTs & args)
     if (args.size() < 2 || args.size() > 3)
         throw Exception("Args not matched, should be: query, database-name[, table-name]", ErrorCodes::BAD_ARGUMENTS);
 
-    String query = safeGet<String>(typeid_cast<const ASTLiteral &>(*args[0]).value);
+    auto query = safeGet<String>(typeid_cast<const ASTLiteral &>(*args[0]).value);
     const String & database_name = typeid_cast<const ASTIdentifier &>(*args[1]).name;
 
     if (args.size() == 3)

dbms/src/Flash/BatchCoprocessorHandler.cpp

@@ -18,8 +18,8 @@
 #include <Flash/Coprocessor/InterpreterDAG.h>
 #include <Flash/ServiceUtils.h>
 #include <Storages/IStorage.h>
-#include <Storages/Transaction/SchemaSyncer.h>
 #include <Storages/Transaction/TMTContext.h>
+#include <TiDB/Schema/SchemaSyncer.h>
 
 #include <ext/scope_guard.h>
 

dbms/src/Flash/Coprocessor/DAGStorageInterpreter.cpp

@@ -33,7 +33,7 @@
 #include <Storages/MutableSupport.h>
 #include <Storages/Transaction/KVStore.h>
 #include <Storages/Transaction/LockException.h>
-#include <Storages/Transaction/SchemaSyncer.h>
+#include <TiDB/Schema/SchemaSyncer.h>
 
 #pragma GCC diagnostic push
 #pragma GCC diagnostic ignored "-Wunused-parameter"

dbms/src/Flash/CoprocessorHandler.cpp

@@ -22,8 +22,8 @@
 #include <Storages/IStorage.h>
 #include <Storages/Transaction/LockException.h>
 #include <Storages/Transaction/RegionException.h>
-#include <Storages/Transaction/SchemaSyncer.h>
 #include <Storages/Transaction/TMTContext.h>
+#include <TiDB/Schema/SchemaSyncer.h>
 
 #include <ext/scope_guard.h>
 

dbms/src/Interpreters/Context.cpp

@@ -59,9 +59,9 @@
 #include <Storages/PathCapacityMetrics.h>
 #include <Storages/PathPool.h>
 #include <Storages/Transaction/BackgroundService.h>
-#include <Storages/Transaction/SchemaSyncService.h>
 #include <Storages/Transaction/TMTContext.h>
 #include <TableFunctions/TableFunctionFactory.h>
+#include <TiDB/Schema/SchemaSyncService.h>
 #include <common/logger_useful.h>
 #include <fiu.h>
 #include <fmt/core.h>
@@ -207,6 +207,7 @@ struct ContextShared
 
     explicit ContextShared(std::shared_ptr<IRuntimeComponentsFactory> runtime_components_factory_)
         : runtime_components_factory(std::move(runtime_components_factory_))
+        , storage_run_mode(PageStorageRunMode::ONLY_V3)
     {
         /// TODO: make it singleton (?)
         static std::atomic<size_t> num_calls{0};

dbms/src/Interpreters/IDAsPathUpgrader.cpp

@@ -33,10 +33,10 @@
 #include <Poco/FileStream.h>
 #include <Storages/MutableSupport.h>
 #include <Storages/PathPool.h>
-#include <Storages/Transaction/SchemaNameMapper.h>
 #include <Storages/Transaction/TMTContext.h>
 #include <Storages/Transaction/TiDB.h>
-#include <Storages/Transaction/TiDBSchemaSyncer.h>
+#include <TiDB/Schema/SchemaNameMapper.h>
+#include <TiDB/Schema/TiDBSchemaSyncer.h>
 #include <common/logger_useful.h>
 #include <fmt/core.h>
 
@@ -71,7 +71,7 @@ std::shared_ptr<ASTFunction> getDatabaseEngine(const FileProviderPtr & file_prov
     ParserCreateQuery parser;
     ASTPtr ast = parseQuery(parser, query.data(), query.data() + query.size(), "in file " + filename, 0);
     ASTCreateQuery & ast_create_query = typeid_cast<ASTCreateQuery &>(*ast);
-    auto storage = ast_create_query.storage;
+    auto * storage = ast_create_query.storage;
     if (storage == nullptr || storage->engine == nullptr || storage->engine->name.empty())
     {
         throw Exception("Can not get database engine for file: " + filename, ErrorCodes::LOGICAL_ERROR);
@@ -97,15 +97,15 @@ std::pair<String, TiDB::TableInfo> getTableInfo(const FileProviderPtr & file_pro
     ParserCreateQuery parser;
     ASTPtr ast = parseQuery(parser, definition.data(), definition.data() + definition.size(), "in file " + table_metadata_file, 0);
     ASTCreateQuery & ast_create_query = typeid_cast<ASTCreateQuery &>(*ast);
-    auto storage = ast_create_query.storage;
+    auto * storage = ast_create_query.storage;
     if (storage == nullptr || storage->engine == nullptr || storage->engine->name.empty())
     {
         throw Exception("Can not get table engine for file: " + table_metadata_file, ErrorCodes::LOGICAL_ERROR);
     }
 
     TiDB::TableInfo info;
     ASTFunction * engine = storage->engine;
-    auto * args = typeid_cast<const ASTExpressionList *>(engine->arguments.get());
+    const auto * args = typeid_cast<const ASTExpressionList *>(engine->arguments.get());
     if (args == nullptr)
         throw Exception("Can not cast table engine arguments", ErrorCodes::BAD_ARGUMENTS);
 
@@ -399,12 +399,12 @@ String IDAsPathUpgrader::DatabaseDiskInfo::getNewMetaDirectory(const String & ro
     return root_path + (endsWith(root_path, "/") ? "" : "/") + "/metadata/" + escapeForFileName(newName()) + "/";
 }
 // "data/"
-String IDAsPathUpgrader::DatabaseDiskInfo::getNewDataDirectory(const String & root_path) const
+String IDAsPathUpgrader::DatabaseDiskInfo::getNewDataDirectory(const String & root_path)
 {
     return root_path + "/data/";
 }
 // "extra_data/"
-String IDAsPathUpgrader::DatabaseDiskInfo::getNewExtraDirectory(const String & extra_root) const
+String IDAsPathUpgrader::DatabaseDiskInfo::getNewExtraDirectory(const String & extra_root)
 {
     return extra_root + "/";
 }
@@ -457,11 +457,11 @@ IDAsPathUpgrader::IDAsPathUpgrader(Context & global_ctx_, bool is_mock_, std::un
 
 bool IDAsPathUpgrader::needUpgrade()
 {
-    const auto metadataPath = global_context.getPath() + "/metadata";
+    const auto metadata_path = global_context.getPath() + "/metadata";
 
     // For old version, we have database directories and its `.sql` file
     Poco::DirectoryIterator dir_end;
-    for (Poco::DirectoryIterator it(metadataPath); it != dir_end; ++it)
+    for (Poco::DirectoryIterator it(metadata_path); it != dir_end; ++it)
     {
         if (!it->isDirectory())
             continue;
@@ -893,7 +893,7 @@ void IDAsPathUpgrader::renameTable(
                     args->children.emplace_back(literal);
                 else if (args->children.size() >= 2)
                     args->children.at(1) = literal;
-            } while (0);
+            } while (false);
         }
 
         const String new_tbl_meta_file = table.getNewMetaFilePath(root_path, db_info);

dbms/src/Interpreters/IDAsPathUpgrader.h

@@ -137,9 +137,9 @@ class IDAsPathUpgrader
         // "metadata/db_${id}/"
         String getNewMetaDirectory(const String & root_path) const;
         // "data/"
-        String getNewDataDirectory(const String & root_path) const;
+        static String getNewDataDirectory(const String & root_path);
         // "extra_data/"
-        String getNewExtraDirectory(const String & extra_root) const;
+        static String getNewExtraDirectory(const String & extra_root);
 
     private:
         // "metadata/${db_name}.sql"

dbms/src/Interpreters/InterpreterSelectQuery.cpp

@@ -55,12 +55,12 @@
 #include <Storages/RegionQueryInfo.h>
 #include <Storages/Transaction/LearnerRead.h>
 #include <Storages/Transaction/RegionRangeKeys.h>
-#include <Storages/Transaction/SchemaSyncer.h>
 #include <Storages/Transaction/StorageEngineType.h>
 #include <Storages/Transaction/TMTContext.h>
 #include <Storages/Transaction/TiKVRange.h>
 #include <TableFunctions/ITableFunction.h>
 #include <TableFunctions/TableFunctionFactory.h>
+#include <TiDB/Schema/SchemaSyncer.h>
 
 #pragma GCC diagnostic push
 #pragma GCC diagnostic ignored "-Wunused-parameter"
@@ -496,7 +496,7 @@ void InterpreterSelectQuery::executeImpl(Pipeline & pipeline, const BlockInputSt
         {
             if (expressions.has_join)
             {
-                const ASTTableJoin & join = static_cast<const ASTTableJoin &>(*query.join()->table_join);
+                const auto & join = static_cast<const ASTTableJoin &>(*query.join()->table_join);
                 if (join.kind == ASTTableJoin::Kind::Full || join.kind == ASTTableJoin::Kind::Right)
                     pipeline.stream_with_non_joined_data = expressions.before_join->createStreamWithNonJoinedDataIfFullOrRightJoin(
                         pipeline.firstStream()->getHeader(),
@@ -816,7 +816,7 @@ QueryProcessingStage::Enum InterpreterSelectQuery::executeFetchColumns(Pipeline
 
             for (size_t i = 0; i < arr->size(); i++)
             {
-                String str = arr->getElement<String>(i);
+                auto str = arr->getElement<String>(i);
                 ::metapb::Region region;
                 ::google::protobuf::TextFormat::ParseFromString(str, &region);
 
@@ -839,7 +839,7 @@ QueryProcessingStage::Enum InterpreterSelectQuery::executeFetchColumns(Pipeline
         }
 
         /// PARTITION SELECT only supports MergeTree family now.
-        if (const ASTSelectQuery * select_query = typeid_cast<const ASTSelectQuery *>(query_info.query.get()))
+        if (const auto * select_query = typeid_cast<const ASTSelectQuery *>(query_info.query.get()))
         {
             if (select_query->partition_expression_list)
             {
@@ -860,7 +860,7 @@ QueryProcessingStage::Enum InterpreterSelectQuery::executeFetchColumns(Pipeline
             if (auto managed_storage = std::dynamic_pointer_cast<IManageableStorage>(storage);
                 managed_storage && managed_storage->engineType() == TiDB::StorageEngine::DT)
             {
-                if (const ASTSelectQuery * select_query = typeid_cast<const ASTSelectQuery *>(query_info.query.get()))
+                if (const auto * select_query = typeid_cast<const ASTSelectQuery *>(query_info.query.get()))
                 {
                     // With `no_kvsotre` is true, we do not do learner read
                     if (likely(!select_query->no_kvstore))
@@ -910,7 +910,7 @@ QueryProcessingStage::Enum InterpreterSelectQuery::executeFetchColumns(Pipeline
             QuotaForIntervals & quota = context.getQuota();
 
             pipeline.transform([&](auto & stream) {
-                if (IProfilingBlockInputStream * p_stream = dynamic_cast<IProfilingBlockInputStream *>(stream.get()))
+                if (auto * p_stream = dynamic_cast<IProfilingBlockInputStream *>(stream.get()))
                 {
                     p_stream->setLimits(limits);
 
@@ -1275,7 +1275,7 @@ void InterpreterSelectQuery::executeLimitBy(Pipeline & pipeline) // NOLINT
     for (const auto & elem : query.limit_by_expression_list->children)
         columns.emplace_back(elem->getColumnName());
 
-    size_t value = safeGet<UInt64>(typeid_cast<ASTLiteral &>(*query.limit_by_value).value);
+    auto value = safeGet<UInt64>(typeid_cast<ASTLiteral &>(*query.limit_by_value).value);
 
     pipeline.transform([&](auto & stream) {
         stream = std::make_shared<LimitByBlockInputStream>(stream, value, columns);
@@ -1347,7 +1347,7 @@ void InterpreterSelectQuery::executeExtremes(Pipeline & pipeline)
         return;
 
     pipeline.transform([&](auto & stream) {
-        if (IProfilingBlockInputStream * p_stream = dynamic_cast<IProfilingBlockInputStream *>(stream.get()))
+        if (auto * p_stream = dynamic_cast<IProfilingBlockInputStream *>(stream.get()))
             p_stream->enableExtremes();
     });
 }