Fixed race condition in DROP/CREATE MergeTree tables [#CLICKHOUSE-3939]

dbms/src/Common/CurrentThread.h

@@ -82,7 +82,6 @@ public:
         ~QueryScope();
     };
 
-public:
     /// Implicitly finalizes current thread in the destructor
     class ThreadScope
     {

dbms/src/Interpreters/InterpreterCreateQuery.cpp

@@ -576,9 +576,18 @@ BlockIO InterpreterCreateQuery::createTable(ASTCreateQuery & create)
             context.getSessionContext().addExternalTable(table_name, res, query_ptr);
         else
             database->createTable(context, table_name, res, query_ptr);
-    }
 
-    res->startup();
+        /// We must call "startup" and "shutdown" while holding DDLGuard.
+        /// Because otherwise method "shutdown" (from InterpreterDropQuery) can be called before startup
+        /// (in case when table was created and instantly dropped before started up)
+        ///
+        /// Method "startup" may create background tasks and method "shutdown" will wait for them.
+        /// But if "shutdown" is called before "startup", it will exit early, because there are no background tasks to wait.
+        /// Then background task is created by "startup" method. And when destructor of a table object is called, background task is still active,
+        /// and the task will use references to freed data.
+
+        res->startup();
+    }
 
     /// If the query is a CREATE SELECT, insert the data into the table.
     if (create.select && !create.attach

dbms/src/Interpreters/InterpreterDropQuery.cpp

@@ -208,7 +208,7 @@ DatabaseAndTable InterpreterDropQuery::tryGetDatabaseAndTable(String & database_
             throw Exception("Table " + backQuoteIfNeed(database_name) + "." + backQuoteIfNeed(table_name) + " doesn't exist.",
                             ErrorCodes::UNKNOWN_TABLE);
 
-        return std::make_pair<DatabasePtr, StoragePtr>(std::move(database), std::move(table));
+        return {std::move(database), std::move(table)};
     }
     return {};
 }

dbms/tests/queries/0_stateless/00705_drop_create_merge_tree.reference

@@ -0,0 +1 @@
+

dbms/tests/queries/0_stateless/00705_drop_create_merge_tree.sh

@@ -0,0 +1,25 @@
+#!/usr/bin/env bash
+set -e
+
+CURDIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)
+. $CURDIR/../shell_config.sh
+
+function stress()
+{
+    while true; do
+        ${CLICKHOUSE_CLIENT} --query "CREATE TABLE IF NOT EXISTS test.table (x UInt8) ENGINE = MergeTree ORDER BY tuple()" 2>/dev/null
+        ${CLICKHOUSE_CLIENT} --query "DROP TABLE test.table" 2>/dev/null
+    done
+}
+
+# https://stackoverflow.com/questions/9954794/execute-a-shell-function-with-timeout
+export -f stress
+
+for thread in {1..5}; do
+    timeout 10 bash -c stress &
+done
+
+wait
+echo
+
+${CLICKHOUSE_CLIENT} --query "DROP TABLE IF EXISTS test.table";