[DDL] Pass all txn tests (#116)

* Add sync schema on read

* Simplify schema syncer interface and adjust mock stuff

* Rename default schema version setting

* Compensate last commit

* Remove curl library

* Remove curl from builder image

* Remove useless codes, init schema syncer based on pd config

* Minor fix to schema debug

* Fix alter tmt and pass tests

* Fix build fail

* Add lock for mock schema syncer

* Fix schema sync service init context

* Adjust schema tests

* Not sync if no schema change detected

* Adjust txn mock tests

dbms/src/Debug/MockSchemaSyncer.cpp

@@ -236,6 +236,8 @@ MockSchemaSyncer::MockSchemaSyncer() : log(&Logger::get("MockSchemaSyncer")) {}
 
 bool MockSchemaSyncer::syncSchemas(Context & context)
 {
+    std::lock_guard<std::mutex> lock(schema_mutex);
+
     std::unordered_map<TableID, MockTiDB::TablePtr> new_tables;
     MockTiDB::instance().traverseTables([&](const auto & table) { new_tables.emplace(table->id(), table); });
 
@@ -261,7 +263,7 @@ void MockSchemaSyncer::syncTable(Context & context, MockTiDB::TablePtr table)
     auto & tmt_context = context.getTMTContext();
 
     /// Get table schema json.
-    TableInfo table_info = table->table_info;
+    const TableInfo & table_info = table->table_info;
     auto table_id = table_info.id;
 
     auto storage = tmt_context.getStorages().get(table_id);
@@ -278,17 +280,16 @@ void MockSchemaSyncer::syncTable(Context & context, MockTiDB::TablePtr table)
         auto create_table_internal = [&]() {
             LOG_DEBUG(log, __PRETTY_FUNCTION__ << ": Creating table " << table_info.name);
             createTable(table_info, context);
-            auto logical_storage = std::static_pointer_cast<StorageMergeTree>(context.getTable(table_info.db_name, table_info.name));
-            context.getTMTContext().getStorages().put(logical_storage);
 
-            /// Mangle for partition table.
-            bool is_partition_table = table_info.manglePartitionTableIfNeeded(table_id);
-            if (is_partition_table && !context.isTableExist(table_info.db_name, table_info.name))
+            /// Create sub-table for partitions if any.
+            if (table_info.is_partition_table)
             {
-                LOG_DEBUG(log, __PRETTY_FUNCTION__ << ": Re-creating table after mangling partition table " << table_info.name);
-                createTable(table_info, context);
-                auto physical_storage = std::static_pointer_cast<StorageMergeTree>(context.getTable(table_info.db_name, table_info.name));
-                context.getTMTContext().getStorages().put(physical_storage);
+                // create partition table.
+                for (auto part_def : table_info.partition.definitions)
+                {
+                    auto part_table_info = table_info.producePartitionTableInfo(part_def.id);
+                    createTable(*part_table_info, context);
+                }
             }
         };
 
@@ -343,12 +344,15 @@ void MockSchemaSyncer::syncTable(Context & context, MockTiDB::TablePtr table)
 
     LOG_DEBUG(log, __PRETTY_FUNCTION__ << ": " << ss.str());
 
-    // Call storage alter to apply schema changes.
-    storage->alterForTMT(alter_commands, table_info, table->table_info.db_name, context);
+    if (!alter_commands.empty())
+    {
+        // Call storage alter to apply schema changes.
+        storage->alterForTMT(alter_commands, table_info, table->table_info.db_name, context);
 
-    LOG_DEBUG(log, __PRETTY_FUNCTION__ << ": Schema changes apply done.");
+        LOG_DEBUG(log, __PRETTY_FUNCTION__ << ": Schema changes apply done.");
 
-    // TODO: Apply schema changes to partition tables.
+        // TODO: Apply schema changes to partition tables.
+    }
 }
 
 } // namespace DB

dbms/src/Debug/MockSchemaSyncer.h

@@ -19,6 +19,8 @@ class MockSchemaSyncer : public SchemaSyncer
 
     Logger * log;
 
+    std::mutex schema_mutex;
+
     std::unordered_map<TableID, MockTiDB::TablePtr> tables;
 };
 

dbms/src/Interpreters/Context.cpp

@@ -1460,7 +1460,7 @@ void Context::initializeSchemaSyncService()
     auto lock = getLock();
     if (shared->schema_sync_service)
         throw Exception("Schema Sync Service has already been initialized.", ErrorCodes::LOGICAL_ERROR);
-    shared->schema_sync_service = std::make_shared<SchemaSyncService>(*this);
+    shared->schema_sync_service = std::make_shared<SchemaSyncService>(*global_context);
 }
 
 SchemaSyncServicePtr & Context::getSchemaSyncService()

tests/mutable-test/txn_mock/data_only_in_region.test

@@ -1,5 +1,5 @@
-=> drop table if exists default.test
 => DBGInvoke __drop_tidb_table(default, test)
+=> drop table if exists default.test
 
 => DBGInvoke __set_flush_threshold(100000, 20000000)
 => DBGInvoke mock_schema_syncer('true')
@@ -9,7 +9,6 @@
 
 => DBGInvoke __mock_tidb_table(default, test, 'col_1 String, col_2 Int64')
 => DBGInvoke __put_region(4, 0, 100, default, test)
-=> DBGInvoke __refresh_schema(default, test)
 
 => DBGInvoke __raft_insert_row(default, test, 4, 50, 'test1', -1)
 => DBGInvoke __raft_insert_row(default, test, 4, 51, 'test2', -2)
@@ -36,5 +35,5 @@
 │       0 │
 └─────────┘
 
-=> drop table if exists default.test
 => DBGInvoke __drop_tidb_table(default, test)
+=> drop table if exists default.test

tests/mutable-test/txn_mock/delete.test

@@ -1,5 +1,5 @@
-=> drop table if exists default.test
 => DBGInvoke __drop_tidb_table(default, test)
+=> drop table if exists default.test
 
 => DBGInvoke __set_flush_threshold(100000, 0)
 => DBGInvoke mock_schema_syncer('true')
@@ -62,5 +62,5 @@
 │ test3 │     3 │          52 │                 0 │
 └───────┴───────┴─────────────┴───────────────────┘
 
-=> drop table if exists default.test
 => DBGInvoke __drop_tidb_table(default, test)
+=> drop table if exists default.test

tests/mutable-test/txn_mock/insert.test

@@ -1,5 +1,5 @@
-=> drop table if exists default.test
 => DBGInvoke __drop_tidb_table(default, test)
+=> drop table if exists default.test
 
 => DBGInvoke __set_flush_threshold(100000, 0)
 => DBGInvoke mock_schema_syncer('true')
@@ -33,5 +33,5 @@
 │       4 │
 └─────────┘
 
-=> drop table if exists default.test
 => DBGInvoke __drop_tidb_table(default, test)
+=> drop table if exists default.test

tests/mutable-test/txn_mock/order_by.test

@@ -1,5 +1,5 @@
-=> drop table if exists default.test
 => DBGInvoke __drop_tidb_table(default, test)
+=> drop table if exists default.test
 
 => DBGInvoke __set_flush_threshold(100000, 0)
 => DBGInvoke mock_schema_syncer('true')
@@ -33,5 +33,5 @@
 │ test1 │    -1 │          50 │
 └───────┴───────┴─────────────┘
 
-=> drop table if exists default.test
 => DBGInvoke __drop_tidb_table(default, test)
+=> drop table if exists default.test

tests/mutable-test/txn_mock/partition_table.test

@@ -1,22 +1,22 @@
+=> DBGInvoke __enable_schema_sync_service('false')
+
+=> DBGInvoke __drop_tidb_table(default, test)
 => drop table if exists default.test
 => drop table if exists default.test_p1
 => drop table if exists default.test_p2
-=> DBGInvoke __drop_tidb_table(default, test)
 
 => DBGInvoke __mock_schema_syncer('true')
 
 => DBGInvoke __mock_tidb_table(default, test, 'col_1 String, col_2 Int64')
-
 => DBGInvoke __mock_tidb_partition(default, test, p1)
-=> DBGInvoke __put_region(4, 0, 100, default, test, p1)
+=> DBGInvoke __mock_tidb_partition(default, test, p2)
 
+=> DBGInvoke __put_region(4, 0, 100, default, test, p1)
 => DBGInvoke __raft_insert_row(default, test, 4, 50, 'test1', 1)
 => DBGInvoke __raft_insert_row(default, test, 4, 51, 'test2', 2)
 => DBGInvoke __try_flush_region(4)
 
-=> DBGInvoke __mock_tidb_partition(default, test, p2)
 => DBGInvoke __put_region(5, 100, 200, default, test, p2)
-
 => DBGInvoke __raft_insert_row(default, test, 5, 152, 'test3', 3)
 => DBGInvoke __raft_insert_row(default, test, 5, 153, 'test4', 4)
 => DBGInvoke __try_flush_region(5)
@@ -44,7 +44,8 @@
 │       2 │
 └─────────┘
 
+=> DBGInvoke __drop_tidb_table(default, test)
 => drop table if exists default.test
 => drop table if exists default.test_p1
 => drop table if exists default.test_p2
-=> DBGInvoke __drop_tidb_table(default, test)
+=> DBGInvoke __enable_schema_sync_service('true')

tests/mutable-test/txn_mock/region.test

@@ -1,5 +1,5 @@
-=> drop table if exists default.test
 => DBGInvoke __drop_tidb_table(default, test)
+=> drop table if exists default.test
 
 => DBGInvoke __set_flush_threshold(1, 2330)
 => DBGInvoke mock_schema_syncer('true')
@@ -32,5 +32,5 @@
 │   80004 │
 └─────────┘
 
-=> drop table if exists default.test
 => DBGInvoke __drop_tidb_table(default, test)
+=> drop table if exists default.test

tests/mutable-test/txn_mock/same_version_diff_delmark.test

@@ -1,5 +1,5 @@
-=> drop table if exists default.test
 => DBGInvoke __drop_tidb_table(default, test)
+=> drop table if exists default.test
 => DBGInvoke mock_schema_syncer('true')
 ┌─mock_schema_syncer("true")─┐
 │ mock schema syncer enabled │
@@ -36,5 +36,5 @@
 │    12 │           3 │                 4 │                 0 │
 │     0 │           3 │                 4 │                 1 │
 └───────┴─────────────┴───────────────────┴───────────────────┘
-=> drop table if exists default.test
 => DBGInvoke __drop_tidb_table(default, test)
+=> drop table if exists default.test

tests/mutable-test/txn_mock/select.test

@@ -1,5 +1,5 @@
-=> drop table if exists default.test
 => DBGInvoke __drop_tidb_table(default, test)
+=> drop table if exists default.test
 
 => DBGInvoke __set_flush_threshold(100000, 0)
 => DBGInvoke mock_schema_syncer('true')
@@ -41,5 +41,5 @@
 │          2 │ test2 │
 └────────────┴───────┘
 
-=> drop table if exists default.test
 => DBGInvoke __drop_tidb_table(default, test)
+=> drop table if exists default.test

tests/mutable-test/txn_mock/selraw.test

@@ -1,5 +1,5 @@
-=> drop table if exists default.test
 => DBGInvoke __drop_tidb_table(default, test)
+=> drop table if exists default.test
 
 => DBGInvoke __set_flush_threshold(100000, 0)
 => DBGInvoke mock_schema_syncer('true')
@@ -42,5 +42,5 @@
 │          2 │ test2 │
 └────────────┴───────┘
 
-=> drop table if exists default.test
 => DBGInvoke __drop_tidb_table(default, test)
+=> drop table if exists default.test

tests/mutable-test/txn_mock/snapshot.test

@@ -1,5 +1,5 @@
-=> drop table if exists default.test
 => DBGInvoke __drop_tidb_table(default, test)
+=> drop table if exists default.test
 => DBGInvoke mock_schema_syncer('true')
 ┌─mock_schema_syncer("true")─┐
 │ mock schema syncer enabled │
@@ -55,5 +55,5 @@
 │    13 │           1 │                 3 │                 0 │
 │    11 │           2 │                 3 │                 0 │
 └───────┴─────────────┴───────────────────┴───────────────────┘
-=> drop table if exists default.test
 => DBGInvoke __drop_tidb_table(default, test)
+=> drop table if exists default.test

tests/mutable-test/txn_mock/snapshot_cache.test

@@ -1,5 +1,5 @@
-=> drop table if exists default.test
 => DBGInvoke __drop_tidb_table(default, test)
+=> drop table if exists default.test
 => DBGInvoke mock_schema_syncer('true')
 ┌─mock_schema_syncer("true")─┐
 │ mock schema syncer enabled │
@@ -54,5 +54,5 @@
 │    19 │           2 │
 │    20 │           4 │
 └───────┴─────────────┘
-=> drop table if exists default.test
 => DBGInvoke __drop_tidb_table(default, test)
+=> drop table if exists default.test

tests/mutable-test/txn_schema/alter.test

@@ -1,4 +1,6 @@
 # Preparation.
+=> DBGInvoke __enable_schema_sync_service('false')
+
 => DBGInvoke __drop_tidb_table(default, test)
 => drop table if exists default.test
 
@@ -134,3 +136,4 @@ Code: 47. DB::Exception: Received from {#WORD} DB::Exception: Unknown identifier
 # Clean up.
 => DBGInvoke __drop_tidb_table(default, test)
 => drop table if exists default.test
+=> DBGInvoke __enable_schema_sync_service('true')

tests/mutable-test/txn_schema/drop.test

@@ -1,3 +1,5 @@
+=> DBGInvoke __enable_schema_sync_service('false')
+
 => DBGInvoke __drop_tidb_table(default, test)
 => drop table if exists default.test
 
@@ -40,3 +42,4 @@ Code: 60. DB::Exception: Received from {#WORD} DB::Exception: Table default.test
 
 => DBGInvoke __drop_tidb_table(default, test)
 => drop table if exists default.test
+=> DBGInvoke __enable_schema_sync_service('true')