From 26b6dd7925d3725f401abd093197c651cc148967 Mon Sep 17 00:00:00 2001
From: zanmato1984 <zanmato1984@gmail.com>
Date: Tue, 16 Jul 2019 17:20:42 +0800
Subject: [PATCH 1/4] Add sync schema on read

---
 dbms/src/Debug/MockSchemaSyncer.h             |  2 +-
 .../Interpreters/InterpreterSelectQuery.cpp   | 57 +++++++++++++------
 .../src/Interpreters/InterpreterSelectQuery.h |  2 +-
 3 files changed, 43 insertions(+), 18 deletions(-)

diff --git a/dbms/src/Debug/MockSchemaSyncer.h b/dbms/src/Debug/MockSchemaSyncer.h
index 5c2dd58b592..52f7d22ac5a 100644
--- a/dbms/src/Debug/MockSchemaSyncer.h
+++ b/dbms/src/Debug/MockSchemaSyncer.h
@@ -12,7 +12,7 @@ class MockSchemaSyncer : public SchemaSyncer
 
     bool syncSchemas(Context & context) override;
 
-    void syncSchema(Context & context, TableID table_id, bool) override;
+    void syncSchema(Context & context, TableID table_id, bool lock) override;
 
     TableID getTableIdByName(const std::string & database_name, const std::string & table_name)
     {
diff --git a/dbms/src/Interpreters/InterpreterSelectQuery.cpp b/dbms/src/Interpreters/InterpreterSelectQuery.cpp
index c312e00422a..2446554e3df 100644
--- a/dbms/src/Interpreters/InterpreterSelectQuery.cpp
+++ b/dbms/src/Interpreters/InterpreterSelectQuery.cpp
@@ -78,6 +78,7 @@ namespace ErrorCodes
     extern const int LOGICAL_ERROR;
     extern const int NOT_IMPLEMENTED;
     extern const int SCHEMA_VERSION_ERROR;
+    extern const int UNKNOWN_EXCEPTION;
 }
 
 InterpreterSelectQuery::InterpreterSelectQuery(
@@ -162,10 +163,7 @@ void InterpreterSelectQuery::init(const Names & required_result_column_names)
     }
 
     if (storage)
-        table_lock = storage->lockStructure(false, __PRETTY_FUNCTION__);
-
-    /// Make sure TMT storage schema qualifies the version specified by upper (TiDB or TiSpark).
-    alignStorageSchema(settings.schema_version);
+        table_lock = alignStorageSchemaAndLock(settings.schema_version);
 
     query_analyzer = std::make_unique<ExpressionAnalyzer>(
         query_ptr, context, storage, source_columns, required_result_column_names, subquery_depth, !only_analyze);
@@ -189,24 +187,51 @@ void InterpreterSelectQuery::init(const Names & required_result_column_names)
 }
 
 
-void InterpreterSelectQuery::alignStorageSchema(Int64 schema_version)
+TableStructureReadLockPtr InterpreterSelectQuery::alignStorageSchemaAndLock(Int64 schema_version)
 {
-    if (schema_version == DEFAULT_SCHEMA_VERSION || !storage)
-        return;
-
+    /// Regular read lock for non-TMT or DEFAULT_SCHEMA_VERSION specified.
     const auto merge_tree = dynamic_cast<const StorageMergeTree *>(storage.get());
-    if (!merge_tree || merge_tree->getData().merging_params.mode != MergeTreeData::MergingParams::Txn)
-        return;
+    if (schema_version == DEFAULT_SCHEMA_VERSION || !merge_tree || merge_tree->getData().merging_params.mode != MergeTreeData::MergingParams::Txn)
+        return storage->lockStructure(false, __PRETTY_FUNCTION__);
+
+    /// Lambda for schema version check under the read lock.
+    auto checkSchemaVersionAndLock = [&](bool schema_synced) -> std::tuple<TableStructureReadLockPtr, Int64> {
+        auto lock = storage->lockStructure(false, __PRETTY_FUNCTION__);
+
+        auto storage_schema_version = merge_tree->getTableInfo().schema_version;
+        if (storage_schema_version > schema_version)
+            throw Exception("Storage schema version " + std::to_string(storage_schema_version) + " newer than query schema version " + std::to_string(schema_version), ErrorCodes::SCHEMA_VERSION_ERROR);
+
+        if ((schema_synced && storage_schema_version <= schema_version) || (!schema_synced && storage_schema_version == schema_version))
+            return std::make_tuple(lock, storage_schema_version);
+
+        return std::make_tuple(nullptr, storage_schema_version);
+    };
 
-    auto storage_schema_version = merge_tree->getTableInfo().schema_version;
-    if (storage_schema_version < schema_version)
+    /// Try check and lock once.
     {
-        LOG_TRACE(log, __PRETTY_FUNCTION__ << " storage schema version: " << storage_schema_version << ", query schema version: " << schema_version << ", syncing schema.");
-        context.getTMTContext().getSchemaSyncer()->syncSchema(context, merge_tree->getTableInfo().id, false);
+        auto [lock, storage_schema_version] = checkSchemaVersionAndLock(false);
+        if (lock)
+        {
+            LOG_DEBUG(log, __PRETTY_FUNCTION__ << " storage schema version: " << storage_schema_version << ", query schema version: " << schema_version << ", schema check OK, no syncing required.");
+            return lock;
+        }
+        LOG_DEBUG(log, __PRETTY_FUNCTION__ << " storage schema version: " << storage_schema_version << ", query schema version: " << schema_version << ", schema check not OK.");
     }
 
-    if (storage_schema_version > schema_version)
-        throw Exception("Storage schema version " + std::to_string(storage_schema_version) + " newer than query schema version " + std::to_string(schema_version), ErrorCodes::SCHEMA_VERSION_ERROR);
+    /// If first try failed, sync schema and check again.
+    {
+        context.getTMTContext().getSchemaSyncer()->syncSchemas(context);
+
+        auto [lock, storage_schema_version] = checkSchemaVersionAndLock(true);
+        if (lock)
+        {
+            LOG_DEBUG(log, __PRETTY_FUNCTION__ << " storage schema version: " << storage_schema_version << ", query schema version: " << schema_version << ", schema check OK after syncing.");
+            return lock;
+        }
+
+        throw Exception("Shouldn't reach here", ErrorCodes::UNKNOWN_EXCEPTION);
+    }
 }
 
 
diff --git a/dbms/src/Interpreters/InterpreterSelectQuery.h b/dbms/src/Interpreters/InterpreterSelectQuery.h
index 24ca7ce350a..c0b007690d0 100644
--- a/dbms/src/Interpreters/InterpreterSelectQuery.h
+++ b/dbms/src/Interpreters/InterpreterSelectQuery.h
@@ -111,7 +111,7 @@ class InterpreterSelectQuery : public IInterpreter
 
     void init(const Names & required_result_column_names);
 
-    void alignStorageSchema(Int64 schema_version);
+    TableStructureReadLockPtr alignStorageSchemaAndLock(Int64 schema_version);
 
     void executeImpl(Pipeline & pipeline, const BlockInputStreamPtr & input, bool dry_run);
 

From d1ac4f42143680cbba7376bb9f03a0b57d6095b7 Mon Sep 17 00:00:00 2001
From: zanmato1984 <zanmato1984@gmail.com>
Date: Tue, 16 Jul 2019 19:52:35 +0800
Subject: [PATCH 2/4] Simplify schema syncer interface and adjust mock stuff

---
 dbms/src/Debug/DBGInvoker.cpp                 |  3 +-
 dbms/src/Debug/MockSchemaSyncer.cpp           | 55 ++++++-----
 dbms/src/Debug/MockSchemaSyncer.h             | 13 +--
 dbms/src/Debug/MockTiDB.cpp                   | 34 ++-----
 dbms/src/Debug/MockTiDB.h                     |  6 +-
 dbms/src/Debug/dbgFuncSchema.cpp              | 92 +++++--------------
 dbms/src/Debug/dbgFuncSchema.h                | 11 ++-
 dbms/src/Interpreters/Context.cpp             |  4 +-
 dbms/src/Interpreters/Context.h               |  2 +-
 .../Storages/Transaction/SchemaSyncService.h  |  2 +
 dbms/src/Storages/Transaction/SchemaSyncer.h  |  9 --
 .../Storages/Transaction/TiDBSchemaSyncer.h   |  2 -
 12 files changed, 78 insertions(+), 155 deletions(-)

diff --git a/dbms/src/Debug/DBGInvoker.cpp b/dbms/src/Debug/DBGInvoker.cpp
index 0f7b97c11bb..ce80ba28ab1 100644
--- a/dbms/src/Debug/DBGInvoker.cpp
+++ b/dbms/src/Debug/DBGInvoker.cpp
@@ -59,8 +59,9 @@ DBGInvoker::DBGInvoker()
 
     regFunc("dump_all_region", dbgFuncDumpAllRegion);
 
+    regFunc("enable_schema_sync_service", dbgFuncEnableSchemaSyncService);
     regFunc("mock_schema_syncer", dbgFuncMockSchemaSyncer);
-    regFunc("refresh_schema", dbgFuncRefreshSchema);
+    regFunc("refresh_schemas", dbgFuncRefreshSchemas);
 }
 
 void replaceSubstr(std::string & str, const std::string & target, const std::string & replacement)
diff --git a/dbms/src/Debug/MockSchemaSyncer.cpp b/dbms/src/Debug/MockSchemaSyncer.cpp
index df41fd98270..116d6295127 100644
--- a/dbms/src/Debug/MockSchemaSyncer.cpp
+++ b/dbms/src/Debug/MockSchemaSyncer.cpp
@@ -235,31 +235,36 @@ AlterCommands detectSchemaChanges(const TableInfo & table_info, const TableInfo
 
 MockSchemaSyncer::MockSchemaSyncer() : log(&Logger::get("MockSchemaSyncer")) {}
 
-bool MockSchemaSyncer::syncSchemas(Context & /*context*/)
+bool MockSchemaSyncer::syncSchemas(Context & context)
 {
-    // Don't do full schema sync, we want to test schema sync timing in a fine-grained fashion.
-    return false;
+    std::unordered_map<TableID, MockTiDB::TablePtr> new_tables;
+    MockTiDB::instance().traverseTables([&](const auto & table) { new_tables.emplace(table->id(), table); });
+
+    for (auto [id, table] : tables)
+    {
+        if (new_tables.find(id) == new_tables.end())
+            dropTable(table->table_info.db_name, table->table_info.name, context);
+    }
+
+    for (auto [id, table] : new_tables)
+    {
+        std::ignore = id;
+        syncTable(context, table);
+    }
+
+    tables.swap(new_tables);
+
+    return true;
 }
 
-void MockSchemaSyncer::syncSchema(Context & context, TableID table_id, bool /*lock*/)
+void MockSchemaSyncer::syncTable(Context & context, MockTiDB::TablePtr table)
 {
     auto & tmt_context = context.getTMTContext();
 
-    /// Get table schema json from TiDB/TiKV.
-    String table_info_json = getSchemaJson(table_id, context);
-    if (table_info_json.empty())
-    {
-        /// Table dropped.
-        auto storage = tmt_context.getStorages().get(table_id);
-        if (storage == nullptr)
-        {
-            LOG_DEBUG(log, __PRETTY_FUNCTION__ << ": Table " << table_id << "doesn't exist in TiDB and doesn't exist in TMT, do nothing.");
-            return;
-        }
-        LOG_DEBUG(log, __PRETTY_FUNCTION__ << ": Table " << table_id << "doesn't exist in TiDB, dropping.");
-        dropTable(storage->getDatabaseName(), storage->getTableName(), context);
-        return;
-    }
+    /// Get table schema json.
+    auto table_id = table->id();
+
+    String table_info_json = table->table_info.serialize(false);
 
     LOG_DEBUG(log, __PRETTY_FUNCTION__ << ": Table " << table_id << " info json: " << table_info_json);
 
@@ -326,8 +331,7 @@ void MockSchemaSyncer::syncSchema(Context & context, TableID table_id, bool /*lo
     }
 
     /// Table existing, detect schema changes and apply.
-    auto merge_tree = std::dynamic_pointer_cast<StorageMergeTree>(storage);
-    const TableInfo & orig_table_info = merge_tree->getTableInfo();
+    const TableInfo & orig_table_info = storage->getTableInfo();
     AlterCommands alter_commands = detectSchemaChanges(table_info, orig_table_info);
 
     std::stringstream ss;
@@ -345,15 +349,8 @@ void MockSchemaSyncer::syncSchema(Context & context, TableID table_id, bool /*lo
 
     LOG_DEBUG(log, __PRETTY_FUNCTION__ << ": " << ss.str());
 
-    {
-        // Change internal TableInfo in TMT first.
-        // TODO: Ideally this should be done within alter function, however we are limited by the narrow alter interface, thus not truly atomic.
-        auto table_hard_lock = storage->lockStructureForAlter(__PRETTY_FUNCTION__);
-        merge_tree->setTableInfo(table_info);
-    }
-
     // Call storage alter to apply schema changes.
-    storage->alter(alter_commands, table_info.db_name, table_info.name, context);
+    storage->alterForTMT(alter_commands, table_info, context);
 
     LOG_DEBUG(log, __PRETTY_FUNCTION__ << ": Schema changes apply done.");
 
diff --git a/dbms/src/Debug/MockSchemaSyncer.h b/dbms/src/Debug/MockSchemaSyncer.h
index 52f7d22ac5a..4139f02c425 100644
--- a/dbms/src/Debug/MockSchemaSyncer.h
+++ b/dbms/src/Debug/MockSchemaSyncer.h
@@ -2,6 +2,8 @@
 
 #include <Debug/MockTiDB.h>
 
+#include <unordered_map>
+
 namespace DB
 {
 
@@ -12,17 +14,12 @@ class MockSchemaSyncer : public SchemaSyncer
 
     bool syncSchemas(Context & context) override;
 
-    void syncSchema(Context & context, TableID table_id, bool lock) override;
-
-    TableID getTableIdByName(const std::string & database_name, const std::string & table_name)
-    {
-        return MockTiDB::instance().getTableIDByName(database_name, table_name);
-    }
-
 protected:
-    String getSchemaJson(TableID table_id, Context & /*context*/) { return MockTiDB::instance().getSchemaJson(table_id); }
+    void syncTable(Context & context, MockTiDB::TablePtr table);
 
     Logger * log;
+
+    std::unordered_map<TableID, MockTiDB::TablePtr> tables;
 };
 
 } // namespace DB
diff --git a/dbms/src/Debug/MockTiDB.cpp b/dbms/src/Debug/MockTiDB.cpp
index 9b937fc1fdc..5ec28ba2598 100644
--- a/dbms/src/Debug/MockTiDB.cpp
+++ b/dbms/src/Debug/MockTiDB.cpp
@@ -26,33 +26,6 @@ Table::Table(const String & database_name_, const String & table_name_, TableInf
     : table_info(std::move(table_info_)), database_name(database_name_), table_name(table_name_)
 {}
 
-String MockTiDB::getSchemaJson(TableID table_id)
-{
-    std::lock_guard lock(tables_mutex);
-
-    auto it = tables_by_id.find(table_id);
-    if (it == tables_by_id.end())
-    {
-        return "";
-    }
-
-    return it->second->table_info.serialize(false);
-}
-
-TableID MockTiDB::getTableIDByName(const std::string & database_name, const std::string & table_name)
-{
-    std::lock_guard lock(tables_mutex);
-
-    String qualified_name = database_name + "." + table_name;
-    auto it = tables_by_name.find(qualified_name);
-    if (it == tables_by_name.end())
-    {
-        return InvalidTableID;
-    }
-
-    return it->second->table_info.id;
-}
-
 void MockTiDB::dropTable(const String & database_name, const String & table_name)
 {
     std::lock_guard lock(tables_mutex);
@@ -251,6 +224,13 @@ TablePtr MockTiDB::getTableByName(const String & database_name, const String & t
     return getTableByNameInternal(database_name, table_name);
 }
 
+void MockTiDB::traverseTables(std::function<void(TablePtr)> f)
+{
+    std::lock_guard lock(tables_mutex);
+
+    std::for_each(tables_by_id.begin(), tables_by_id.end(), [&](const auto & pair) { f(pair.second); });
+}
+
 TablePtr MockTiDB::getTableByNameInternal(const String & database_name, const String & table_name)
 {
     String qualified_name = database_name + "." + table_name;
diff --git a/dbms/src/Debug/MockTiDB.h b/dbms/src/Debug/MockTiDB.h
index 50bd28d7a15..5adc0dce5eb 100644
--- a/dbms/src/Debug/MockTiDB.h
+++ b/dbms/src/Debug/MockTiDB.h
@@ -62,10 +62,6 @@ class MockTiDB : public ext::singleton<MockTiDB>
     using TablePtr = std::shared_ptr<Table>;
 
 public:
-    String getSchemaJson(TableID table_id);
-
-    TableID getTableIDByName(const std::string & database_name, const std::string & table_name);
-
     TableID newTable(const String & database_name, const String & table_name, const ColumnsDescription & columns);
 
     TableID newPartition(const String & database_name, const String & table_name, const String & partition_name);
@@ -80,6 +76,8 @@ class MockTiDB : public ext::singleton<MockTiDB>
 
     TablePtr getTableByName(const String & database_name, const String & table_name);
 
+    void traverseTables(std::function<void(TablePtr)> f);
+
 private:
     TablePtr getTableByNameInternal(const String & database_name, const String & table_name);
 
diff --git a/dbms/src/Debug/dbgFuncSchema.cpp b/dbms/src/Debug/dbgFuncSchema.cpp
index 731ac90899c..dc45d067e76 100644
--- a/dbms/src/Debug/dbgFuncSchema.cpp
+++ b/dbms/src/Debug/dbgFuncSchema.cpp
@@ -3,6 +3,7 @@
 #include <Parsers/ASTIdentifier.h>
 #include <Parsers/ASTLiteral.h>
 #include <Storages/StorageMergeTree.h>
+#include <Storages/Transaction/SchemaSyncService.h>
 #include <Storages/Transaction/SchemaSyncer.h>
 #include <Storages/Transaction/TMTContext.h>
 #include <Storages/Transaction/TiDB.h>
@@ -15,6 +16,23 @@ namespace ErrorCodes
 extern const int UNKNOWN_TABLE;
 } // namespace ErrorCodes
 
+void dbgFuncEnableSchemaSyncService(Context & context, const ASTs & args, DBGInvoker::Printer output)
+{
+    if (args.size() != 1)
+        throw Exception("Args not matched, should be: enable (true/false)", ErrorCodes::BAD_ARGUMENTS);
+
+    bool enable = safeGet<String>(typeid_cast<const ASTLiteral &>(*args[0]).value) == "true";
+
+    if (enable)
+        context.initializeSchemaSyncService();
+    else
+        context.getSchemaSyncService().reset();
+
+    std::stringstream ss;
+    ss << "schema sync service " << (enable ? "enabled" : "disabled");
+    output(ss.str());
+}
+
 void dbgFuncMockSchemaSyncer(Context & context, const ASTs & args, DBGInvoker::Printer output)
 {
     if (args.size() != 1)
@@ -39,79 +57,15 @@ void dbgFuncMockSchemaSyncer(Context & context, const ASTs & args, DBGInvoker::P
     output(ss.str());
 }
 
-void dbgFuncRefreshSchema(Context & context, const ASTs & args, DBGInvoker::Printer output)
+void dbgFuncRefreshSchemas(Context & context, const ASTs &, DBGInvoker::Printer output)
 {
-    if (args.size() != 2)
-        throw Exception("Args not matched, should be: database-name, table-name", ErrorCodes::BAD_ARGUMENTS);
-
-    std::string database_name = typeid_cast<const ASTIdentifier &>(*args[0]).name;
-    std::transform(database_name.begin(), database_name.end(), database_name.begin(), ::tolower);
-    std::string table_name = typeid_cast<const ASTIdentifier &>(*args[1]).name;
-    std::transform(table_name.begin(), table_name.end(), table_name.begin(), ::tolower);
-
-    auto log = [&](TableID table_id) {
-        std::stringstream ss;
-        ss << "refreshed schema for table #" << table_id;
-        output(ss.str());
-    };
-
     TMTContext & tmt = context.getTMTContext();
     auto schema_syncer = tmt.getSchemaSyncer();
-    auto mock_schema_syncer = std::dynamic_pointer_cast<MockSchemaSyncer>(schema_syncer);
-    if (!mock_schema_syncer)
-        throw Exception("Debug function refresh_schema can only be used under mock schema syncer.");
-
-    TableID table_id = mock_schema_syncer->getTableIdByName(database_name, table_name);
-    auto storage = tmt.getStorages().getByName(database_name, table_name);
-
-    if (storage == nullptr && table_id == InvalidTableID)
-        // Table does not exist in CH nor TiDB, error out.
-        throw Exception("Table " + database_name + "." + table_name + " doesn't exist in tidb", ErrorCodes::UNKNOWN_TABLE);
-
-    if (storage == nullptr && table_id != InvalidTableID)
-    {
-        // Table does not exist in CH, but exists in TiDB.
-        // Might be renamed or never synced.
-        // Note there will be a dangling table in CH for the following scenario:
-        // Table t was synced to CH already, then t was renamed (name changed) and truncated (ID changed).
-        // Then this function was called with the new name given, the table will be synced to a new table.
-        // User must manually call this function with the old name to remove the dangling table in CH.
-        mock_schema_syncer->syncSchema(context, table_id, true);
-
-        log(table_id);
-
-        return;
-    }
-
-    if (table_id == InvalidTableID)
-    {
-        // Table exists in CH, but does not exist in TiDB.
-        // Just sync it using the storage's ID, syncer will then remove it.
-        mock_schema_syncer->syncSchema(context, storage->getTableInfo().id, true);
-
-        log(table_id);
-
-        return;
-    }
+    schema_syncer->syncSchemas(context);
 
-    // Table exists in both CH and TiDB.
-    if (table_id != storage->getTableInfo().id)
-    {
-        // Table in TiDB is not the old one, i.e. dropped/renamed then recreated.
-        // Sync the old one in CH first, then sync the new one.
-        mock_schema_syncer->syncSchema(context, storage->getTableInfo().id, true);
-        mock_schema_syncer->syncSchema(context, table_id, true);
-
-        log(table_id);
-
-        return;
-    }
-
-    // Table in TiDB is the same one as in CH.
-    // Just sync it.
-    mock_schema_syncer->syncSchema(context, table_id, true);
-
-    log(table_id);
+    std::stringstream ss;
+    ss << "schemas refreshed";
+    output(ss.str());
 }
 
 } // namespace DB
diff --git a/dbms/src/Debug/dbgFuncSchema.h b/dbms/src/Debug/dbgFuncSchema.h
index 10c56bcf570..9e15ad88ef3 100644
--- a/dbms/src/Debug/dbgFuncSchema.h
+++ b/dbms/src/Debug/dbgFuncSchema.h
@@ -8,14 +8,19 @@ namespace DB
 
 class Context;
 
+// Enable/disable schema sync service.
+// Usage:
+//   ./storages-client.sh "DBGInvoke enable_schema_sync_service(enable)"
+void dbgFuncEnableSchemaSyncService(Context & context, const ASTs & args, DBGInvoker::Printer output);
+
 // Change whether to mock schema syncer.
 // Usage:
 //   ./storages-client.sh "DBGInvoke mock_schema_syncer(enabled)"
 void dbgFuncMockSchemaSyncer(Context & context, const ASTs & args, DBGInvoker::Printer output);
 
-// Refresh schema of the given table.
+// Refresh schemas for all tables.
 // Usage:
-//   ./storage-client.sh "DBGInvoke refresh_schema(database_name, table_name)"
-void dbgFuncRefreshSchema(Context & context, const ASTs & args, DBGInvoker::Printer output);
+//   ./storage-client.sh "DBGInvoke refresh_schemas()"
+void dbgFuncRefreshSchemas(Context & context, const ASTs & args, DBGInvoker::Printer output);
 
 } // namespace DB
diff --git a/dbms/src/Interpreters/Context.cpp b/dbms/src/Interpreters/Context.cpp
index d4ed73d947e..141d611e7bf 100644
--- a/dbms/src/Interpreters/Context.cpp
+++ b/dbms/src/Interpreters/Context.cpp
@@ -1450,12 +1450,12 @@ void Context::initializeSchemaSyncService()
     shared->schema_sync_service = std::make_shared<SchemaSyncService>(*this);
 }
 
-SchemaSyncService & Context::getSchemaSyncService()
+SchemaSyncServicePtr Context::getSchemaSyncService()
 {
     auto lock = getLock();
     if (!shared->schema_sync_service)
         throw Exception("Schema Sync Service is not initialized.", ErrorCodes::LOGICAL_ERROR);
-    return *shared->schema_sync_service;
+    return shared->schema_sync_service;
 }
 
 zkutil::ZooKeeperPtr Context::getZooKeeper() const
diff --git a/dbms/src/Interpreters/Context.h b/dbms/src/Interpreters/Context.h
index c4c53b43d41..4b7402c1a6e 100644
--- a/dbms/src/Interpreters/Context.h
+++ b/dbms/src/Interpreters/Context.h
@@ -369,7 +369,7 @@ class Context
     TiDBService & getTiDBService();
 
     void initializeSchemaSyncService();
-    SchemaSyncService & getSchemaSyncService();
+    SchemaSyncServicePtr getSchemaSyncService();
 
     Clusters & getClusters() const;
     std::shared_ptr<Cluster> getCluster(const std::string & cluster_name) const;
diff --git a/dbms/src/Storages/Transaction/SchemaSyncService.h b/dbms/src/Storages/Transaction/SchemaSyncService.h
index a846f0bf6c5..6a731a98cfb 100644
--- a/dbms/src/Storages/Transaction/SchemaSyncService.h
+++ b/dbms/src/Storages/Transaction/SchemaSyncService.h
@@ -28,4 +28,6 @@ class SchemaSyncService : public std::enable_shared_from_this<SchemaSyncService>
     Logger * log;
 };
 
+using SchemaSyncServicePtr = std::shared_ptr<SchemaSyncService>;
+
 } // namespace DB
diff --git a/dbms/src/Storages/Transaction/SchemaSyncer.h b/dbms/src/Storages/Transaction/SchemaSyncer.h
index 892a5a85839..461c5813853 100644
--- a/dbms/src/Storages/Transaction/SchemaSyncer.h
+++ b/dbms/src/Storages/Transaction/SchemaSyncer.h
@@ -20,15 +20,6 @@ class SchemaSyncer
      * @param context
      */
     virtual bool syncSchemas(Context & context) = 0;
-
-    /**
-     * Synchronize schema between TiDB and CH, to make sure the CH table is new enough to accept data from raft.
-     * Should be stateless.
-     * Nevertheless, the implementations may assume that the storage is appropriately locked, thus still not thread-safe.
-     * @param context
-     * @param table_id
-     */
-    virtual void syncSchema(Context & context, TableID table_id, bool lock) = 0;
 };
 
 using SchemaSyncerPtr = std::shared_ptr<SchemaSyncer>;
diff --git a/dbms/src/Storages/Transaction/TiDBSchemaSyncer.h b/dbms/src/Storages/Transaction/TiDBSchemaSyncer.h
index 79a663b9683..cf6fbb1304c 100644
--- a/dbms/src/Storages/Transaction/TiDBSchemaSyncer.h
+++ b/dbms/src/Storages/Transaction/TiDBSchemaSyncer.h
@@ -52,8 +52,6 @@ struct TiDBSchemaSyncer : public SchemaSyncer
         return true;
     }
 
-    void syncSchema(Context & context, TableID, bool) override { syncSchemas(context); }
-
     bool tryLoadSchemaDiffs(SchemaGetter & getter, Int64 version, Context & context)
     {
         if (isTooOldSchema(cur_version, version))

From d3e2298dd8b98e054545f4fcafcf7aa712cc33b0 Mon Sep 17 00:00:00 2001
From: zanmato1984 <zanmato1984@gmail.com>
Date: Thu, 18 Jul 2019 13:08:55 +0800
Subject: [PATCH 3/4] Rename default schema version setting

---
 dbms/src/Core/Defines.h          | 2 +-
 dbms/src/Interpreters/Settings.h | 2 +-
 2 files changed, 2 insertions(+), 2 deletions(-)

diff --git a/dbms/src/Core/Defines.h b/dbms/src/Core/Defines.h
index 48fe68de6b3..85eb7d499d1 100644
--- a/dbms/src/Core/Defines.h
+++ b/dbms/src/Core/Defines.h
@@ -27,7 +27,7 @@
 #define DEFAULT_MAX_COMPRESS_BLOCK_SIZE 1048576
 
 #define DEFAULT_MAX_READ_TSO 0xFFFFFFFFFFFFFFFF
-#define DEFAULT_SCHEMA_VERSION -1
+#define DEFAULT_UNSPECIFIED_SCHEMA_VERSION -1
 
 /** Which blocks by default read the data (by number of rows).
   * Smaller values give better cache locality, less consumption of RAM, but more overhead to process the query.
diff --git a/dbms/src/Interpreters/Settings.h b/dbms/src/Interpreters/Settings.h
index b6077e0879e..c67999e6916 100644
--- a/dbms/src/Interpreters/Settings.h
+++ b/dbms/src/Interpreters/Settings.h
@@ -29,7 +29,7 @@ struct Settings
     M(SettingString, regions, "", "the region need to be read.") \
     M(SettingBool, resolve_locks, false, "tmt read tso.") \
     M(SettingUInt64, read_tso, DEFAULT_MAX_READ_TSO, "tmt read tso.") \
-    M(SettingInt64, schema_version, DEFAULT_SCHEMA_VERSION, "tmt schema version.") \
+    M(SettingInt64, schema_version, DEFAULT_UNSPECIFIED_SCHEMA_VERSION, "tmt schema version.") \
     M(SettingUInt64, min_compress_block_size, DEFAULT_MIN_COMPRESS_BLOCK_SIZE, "The actual size of the block to compress, if the uncompressed data less than max_compress_block_size is no less than this value and no less than the volume of data for one mark.") \
     M(SettingUInt64, max_compress_block_size, DEFAULT_MAX_COMPRESS_BLOCK_SIZE, "The maximum size of blocks of uncompressed data before compressing for writing to a table.") \
     M(SettingUInt64, max_block_size, DEFAULT_BLOCK_SIZE, "Maximum block size for reading") \

From e690046f1ba4ad7c2f457c061193c7a08f8bfcf8 Mon Sep 17 00:00:00 2001
From: zanmato1984 <zanmato1984@gmail.com>
Date: Thu, 18 Jul 2019 13:15:45 +0800
Subject: [PATCH 4/4] Compensate last commit

---
 dbms/src/Interpreters/InterpreterSelectQuery.cpp | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/dbms/src/Interpreters/InterpreterSelectQuery.cpp b/dbms/src/Interpreters/InterpreterSelectQuery.cpp
index 2446554e3df..b63950e5887 100644
--- a/dbms/src/Interpreters/InterpreterSelectQuery.cpp
+++ b/dbms/src/Interpreters/InterpreterSelectQuery.cpp
@@ -189,9 +189,9 @@ void InterpreterSelectQuery::init(const Names & required_result_column_names)
 
 TableStructureReadLockPtr InterpreterSelectQuery::alignStorageSchemaAndLock(Int64 schema_version)
 {
-    /// Regular read lock for non-TMT or DEFAULT_SCHEMA_VERSION specified.
+    /// Regular read lock for non-TMT or schema version unspecified.
     const auto merge_tree = dynamic_cast<const StorageMergeTree *>(storage.get());
-    if (schema_version == DEFAULT_SCHEMA_VERSION || !merge_tree || merge_tree->getData().merging_params.mode != MergeTreeData::MergingParams::Txn)
+    if (schema_version == DEFAULT_UNSPECIFIED_SCHEMA_VERSION || !merge_tree || merge_tree->getData().merging_params.mode != MergeTreeData::MergingParams::Txn)
         return storage->lockStructure(false, __PRETTY_FUNCTION__);
 
     /// Lambda for schema version check under the read lock.