ddl: Fix sync empty schema diff (#8623)

close #8578

dbms/src/Debug/DBGInvoker.cpp

@@ -90,6 +90,7 @@ DBGInvoker::DBGInvoker()
     regSchemalessFunc("is_tombstone", dbgFuncIsTombstone);
     regSchemalessFunc("refresh_table_schema", dbgFuncRefreshTableSchema);
     regSchemalessFunc("refresh_mapped_table_schema", dbgFuncRefreshMappedTableSchema);
+    regSchemalessFunc("skip_schema_version", dbgFuncSkipSchemaVersion);
 
     regSchemalessFunc("region_split", MockRaftCommand::dbgFuncRegionBatchSplit);
     regSchemalessFunc("region_prepare_merge", MockRaftCommand::dbgFuncPrepareMerge);

dbms/src/Debug/MockTiDB.cpp

@@ -761,6 +761,8 @@ std::pair<bool, DatabaseID> MockTiDB::getDBIDByName(const String & database_name
 
 std::optional<SchemaDiff> MockTiDB::getSchemaDiff(Int64 version_)
 {
+    if (!version_diff.contains(version_))
+        return std::nullopt;
     return version_diff[version_];
 }
 

dbms/src/Debug/MockTiDB.h

@@ -138,6 +138,10 @@ class MockTiDB : public ext::Singleton<MockTiDB>
 
     void truncateTable(const String & database_name, const String & table_name);
 
+    // Mock that concurrent DDL meets conflict, it will retry with a new schema version
+    // Return the schema_version with empty SchemaDiff
+    Int64 skipSchemaVersion() { return ++version; }
+
     TablePtr getTableByName(const String & database_name, const String & table_name);
 
     TiDB::TableInfoPtr getTableInfoByID(TableID table_id);

dbms/src/Debug/dbgFuncSchema.cpp

@@ -29,6 +29,7 @@
 #include <TiDB/Schema/SchemaSyncer.h>
 #include <TiDB/Schema/TiDB.h>
 #include <TiDB/Schema/TiDBSchemaManager.h>
+#include <common/logger_useful.h>
 #include <fmt/core.h>
 
 #include <ext/singleton.h>
@@ -230,5 +231,11 @@ void dbgFuncIsTombstone(Context & context, const ASTs & args, DBGInvoker::Printe
     output(fmt_buf.toString());
 }
 
+void dbgFuncSkipSchemaVersion(Context &, const ASTs &, DBGInvoker::Printer output)
+{
+    auto empty_schema_version = MockTiDB::instance().skipSchemaVersion();
+    LOG_WARNING(Logger::get(), "Generate an empty schema diff with schema_version={}", empty_schema_version);
+    output(fmt::format("Generate an empty schema diff with schema_version={}", empty_schema_version));
+}
 
 } // namespace DB

dbms/src/Debug/dbgFuncSchema.h

@@ -56,4 +56,10 @@ void dbgFuncResetSchemas(Context & context, const ASTs & args, DBGInvoker::Print
 // Usage:
 //   ./storage-client.sh "DBGInvoke is_tombstone(db_name, table_name)"
 void dbgFuncIsTombstone(Context & context, const ASTs & args, DBGInvoker::Printer output);
+
+// Mock that concurrent DDL meets conflict, it will retry the DDL with a new schema version.
+// So the schema_version will contains empty SchemaDiff
+// Usage:
+//   ./storage-client.sh "DBGInvoke skip_schema_version()"
+void dbgFuncSkipSchemaVersion(Context & context, const ASTs & args, DBGInvoker::Printer output);
 } // namespace DB

dbms/src/TiDB/Schema/TiDBSchemaSyncer.cpp

@@ -12,13 +12,14 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+#include <Common/Stopwatch.h>
 #include <Common/TiFlashMetrics.h>
 #include <TiDB/Schema/SchemaBuilder.h>
 #include <TiDB/Schema/TiDBSchemaSyncer.h>
+#include <common/logger_useful.h>
 #include <common/types.h>
 
 #include <mutex>
-#include <shared_mutex>
 
 namespace DB
 {
@@ -88,7 +89,7 @@ bool TiDBSchemaSyncer<mock_getter, mock_mapper>::syncSchemasByGetter(Context & c
             // Since TiDB can not make sure the schema diff of the latest schema version X is not empty, under this situation we should set the `cur_version`
             // to X-1 and try to fetch the schema diff X next time.
             Int64 version_after_load_diff = syncSchemaDiffs(context, getter, version);
-            if (version_after_load_diff != -1)
+            if (version_after_load_diff != SchemaGetter::SchemaVersionNotExist)
             {
                 cur_version = version_after_load_diff;
             }
@@ -114,30 +115,58 @@ Int64 TiDBSchemaSyncer<mock_getter, mock_mapper>::syncSchemaDiffs(
     Getter & getter,
     Int64 latest_version)
 {
-    Int64 used_version = cur_version;
-    // TODO:try to use parallel to speed up
-    while (used_version < latest_version)
-    {
-        used_version++;
-        std::optional<SchemaDiff> diff = getter.getSchemaDiff(used_version);
+    Int64 cur_apply_version = cur_version;
 
-        if (used_version == latest_version && !diff)
+    // If `schema diff` got empty `schema diff`, we should handle it these ways:
+    //
+    // example:
+    //  - `cur_version` is 1, `latest_version` is 10
+    //  - The schema diff of schema version [2,4,6] is empty, Then we just skip it.
+    //  - The schema diff of schema version 10 is empty, Then we should just apply version to 9
+    while (cur_apply_version < latest_version)
+    {
+        cur_apply_version++;
+        std::optional<SchemaDiff> diff = getter.getSchemaDiff(cur_apply_version);
+        if (!diff)
         {
-            --used_version;
+            if (cur_apply_version != latest_version)
+            {
+                // The DDL may meets conflict and the SchemaDiff of `cur_apply_version` is
+                // not used. Skip it.
+                LOG_WARNING(
+                    log,
+                    "Skip an empty schema diff, schema_version={} cur_version={} latest_version={}",
+                    cur_apply_version,
+                    cur_version,
+                    latest_version);
+                continue;
+            }
+
+            assert(cur_apply_version == latest_version);
+            // The latest version is empty, the SchemaDiff may not been committed to TiKV,
+            // skip succeeding the apply version
+            LOG_INFO(
+                log,
+                "Meets an empty schema diff in the latest_version, will not succeed, cur_version={} "
+                "schema_version={}",
+                cur_version,
+                latest_version);
+            --cur_apply_version;
             break;
         }
 
         if (diff->regenerate_schema_map)
         {
-            // If `schema_diff.regenerate_schema_map` == true, return `-1` directly, let TiFlash reload schema info from TiKV.
-            LOG_INFO(log, "Meets a schema diff with regenerate_schema_map flag");
-            return -1;
+            // `FLASHBACK CLUSTER` is executed, return `SchemaGetter::SchemaVersionNotExist`.
+            // The caller should let TiFlash reload schema info from TiKV.
+            LOG_INFO(log, "Meets a schema diff with regenerate_schema_map flag, schema_version={}", cur_apply_version);
+            return SchemaGetter::SchemaVersionNotExist;
         }
 
         SchemaBuilder<Getter, NameMapper> builder(getter, context, databases, table_id_map);
         builder.applyDiff(*diff);
     }
-    return used_version;
+    return cur_apply_version;
 }
 
 template <bool mock_getter, bool mock_mapper>

dbms/src/TiDB/Schema/TiDBSchemaSyncer.h

@@ -15,24 +15,18 @@
 #pragma once
 
 #include <Common/Logger.h>
-#include <Common/Stopwatch.h>
 #include <Debug/MockSchemaGetter.h>
 #include <Debug/MockSchemaNameMapper.h>
 #include <TiDB/Schema/DatabaseInfoCache.h>
 #include <TiDB/Schema/TableIDMap.h>
 #include <TiDB/Schema/TiDB.h>
 #include <pingcap/kv/Cluster.h>
-#include <pingcap/kv/Snapshot.h>
 
 #include <ext/scope_guard.h>
 
 namespace DB
 {
 using KVClusterPtr = std::shared_ptr<pingcap::kv::Cluster>;
-namespace ErrorCodes
-{
-extern const int FAIL_POINT_ERROR;
-};
 
 /// The schema syncer for given keyspace
 template <bool mock_getter, bool mock_mapper>

tests/README.md

@@ -1,13 +1,17 @@
 ## How to run test cases in local
 
-1. Deploy a cluster using [ti.sh](https://github.com/pingcap/tiflash/blob/master/integrated/docs/ti.sh.md)
+For running intergration test cases (defined in `./fullstack-test`, `./fullstack-test-dt`, `./new_collation_fullstack`), you should define a TiDB cluster with TiFlash node (1 PD, 1 TiKV, 1 TiDB, 1 TiFlash at least).
+
+1. Deploy a cluster using [tiup](https://tiup.io/)
 2. Use `ti.sh x.ti prop` to ensure the ports of your cluster
 3. Change the "storage_port", "tidb_port" in `./env.sh` to let it connect to your cluster
 4. Run tests with `./run-test.sh fullstack-test/ddl`
 
-Instead of ti.sh, you can deploy a cluster with [TiUP](https://tiup.io/). But ti.sh is more easy to replace the binary for TiFlash/TiDB/PD, which is convenient for debugging.
-
-* For running mock test cases (defined in `./delta-merge-test`), you should define a cluster with only one TiFlash node, and set standalone property for that node.
-
-* For running intergration test cases (defined in `./fullstack-test`, `./fullstack-test-dt`, `./new_collation_fullstack`), you should define a TiDB cluster with TiFlash node (1 PD, 1 TiKV, 1 TiDB, 1 TiFlash at least).
+## How to run test cases in `delta-merge-test`
 
+For running mock test cases (defined in `./delta-merge-test`), you should start a standalone tiflash that is not connected to tidb cluster
+```
+export TIFLASH_PORT=9000
+./tiflash server -- --path /tmp/tiflash/data --tcp_port ${TIFLASH_PORT}
+storage_port=${TIFLASH_PORT} verbose=true ./run-test.sh delta-merge-test
+```

tests/_env.sh

@@ -54,12 +54,8 @@ export tidb_db="test"
 export tidb_table="t"
 
 # Whether run scripts with verbose output
-export verbose="false"
+export verbose=${verbose:-"false"}
 # export verbose="true"
 
-# Setup running env vars
-#source ../../_vars.sh
-#setup_dylib_path
-
 export LANG=en_US.utf-8
 export LC_ALL=en_US.utf-8

tests/delta-merge-test/raft/schema/concurrent_ddl_conflict.test

@@ -0,0 +1,39 @@
+# Copyright 2023 PingCAP, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+# Preparation.
+=> DBGInvoke __enable_schema_sync_service('false')
+
+=> DBGInvoke __drop_tidb_table(default, test)
+
+=> DBGInvoke __mock_tidb_table(default, test, 'col_1 String', '', 'dt')
+=> DBGInvoke __refresh_mapped_table_schema(default, test)
+=> DBGInvoke __put_region(4, 0, 100, default, test)
+=> DBGInvoke __skip_schema_version()
+=> DBGInvoke __add_column_to_tidb_table(default, test, 'col_2 Nullable(Int8)')
+=> DBGInvoke __add_column_to_tidb_table(default, test, 'col_3 Nullable(Int32)')
+=> DBGInvoke __refresh_schemas()
+=> DBGInvoke __raft_insert_row(default, test, 4, 50, 'test1', 1, 65536)
+
+# Sync add column by reading.
+>> DBGInvoke query_mapped('select col_1,col_2,col_3 from \$d.\$t', default, test)
+┌─col_1─┬─col_2─┬─col_3─┐
+│ test1 │     1 │ 65536 │
+└───────┴───────┴───────┘
+
+# Clean up.
+=> DBGInvoke __drop_tidb_table(default, test)
+=> DBGInvoke __refresh_schemas()
+=> DBGInvoke __enable_schema_sync_service('true')
+

tests/docker/config/tics_dt.toml

@@ -12,13 +12,9 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-tmp_path = "/tmp/tiflash/data/tmp"
-path = "/tmp/tiflash/data/db"
+path = "/tmp/tiflash/data"
 capacity = "107374182400" # 100GB
-mark_cache_size = 1073741824
-minmax_index_cache_size = 1073741824
 tcp_port = 9000
-http_port = 8123
 
 [logger]
 count = 10