diff --git a/dbms/src/Databases/DatabaseAtomic.cpp b/dbms/src/Databases/DatabaseAtomic.cpp
index f66f1a3fcc2bc9a35e5f676064c9d463fae1334e..f09ae2d7d9e96465574da9c869fa506930248d4f 100644
--- a/dbms/src/Databases/DatabaseAtomic.cpp
+++ b/dbms/src/Databases/DatabaseAtomic.cpp
@@ -14,11 +14,13 @@ namespace ErrorCodes
     extern const int FILE_DOESNT_EXIST;
 }
 
-DatabaseAtomic::DatabaseAtomic(String name_, String metadata_path_, const Context & context_)
+DatabaseAtomic::DatabaseAtomic(String name_, String metadata_path_, Context & context_)
     : DatabaseOrdinary(name_, metadata_path_, context_)
 {
     data_path = "store/";
-    log = &Logger::get("DatabaseAtomic (" + name_ + ")");
+    auto log_name = "DatabaseAtomic (" + name_ + ")";
+    log = &Logger::get(log_name);
+    drop_task = context_.getSchedulePool().createTask(log_name, [this](){ this->dropTableDataTask(); });
 }
 
 String DatabaseAtomic::getTableDataPath(const String & table_name) const
@@ -44,6 +46,7 @@ String DatabaseAtomic::getTableDataPath(const ASTCreateQuery & query) const
 void DatabaseAtomic::drop(const Context &)
 {
     Poco::File(getMetadataPath()).remove(false);
+    //TODO check all tables dropped
 }
 
 void DatabaseAtomic::attachTable(const String & name, const StoragePtr & table, const String & relative_table_path)
@@ -63,6 +66,39 @@ StoragePtr DatabaseAtomic::detachTable(const String & name)
     return DatabaseWithDictionaries::detachTable(name);
 }
 
+void DatabaseAtomic::dropTable(const Context & context, const String & table_name)
+{
+    String table_metadata_path = getObjectMetadataPath(table_name);
+    String table_metadata_path_drop = table_metadata_path + drop_suffix;
+    String table_data_path_relative = getTableDataPath(table_name);
+    assert(!table_data_path_relative.empty());
+
+    StoragePtr table = detachTable(table_name);
+    try
+    {
+        // FIXME
+        // 1. CREATE table_name: + table_name.sql
+        // 2. DROP table_name: table_name.sql -> table_name.sql.tmp_drop
+        // 3. CREATE table_name: + table_name.sql
+        // 4. DROP table_name: table_name.sql -> table_name.sql.tmp_drop overwrites table_name.sql.tmp_drop ?
+        Poco::File(table_metadata_path).renameTo(table_metadata_path_drop);
+
+        {
+            LOG_INFO(log, "Mark table " + table->getStorageID().getNameForLogs() + " to drop.");
+            std::lock_guard lock(tables_to_drop_mutex);
+            time_t current_time = std::chrono::system_clock::to_time_t(std::chrono::system_clock::now());
+            tables_to_drop.push_back({table, context.getPath() + table_data_path_relative, current_time});
+        }
+    }
+    catch (...)
+    {
+        LOG_WARNING(log, getCurrentExceptionMessage(__PRETTY_FUNCTION__));
+        attachTable(table_name, table, table_data_path_relative);
+        Poco::File(table_metadata_path_drop).renameTo(table_metadata_path);
+        throw;
+    }
+}
+
 void DatabaseAtomic::renameTable(const Context & context, const String & table_name, IDatabase & to_database,
                                  const String & to_table_name)
 {
@@ -89,6 +125,75 @@ void DatabaseAtomic::renameTable(const Context & context, const String & table_n
     Poco::File(getObjectMetadataPath(table_name)).renameTo(to_database.getObjectMetadataPath(to_table_name));
 }
 
+void DatabaseAtomic::loadStoredObjects(Context & context, bool has_force_restore_data_flag)
+{
+    DatabaseOrdinary::loadStoredObjects(context, has_force_restore_data_flag);
+    drop_task->activateAndSchedule();
+}
+
+void DatabaseAtomic::shutdown()
+{
+    drop_task->deactivate();
+    DatabaseWithDictionaries::shutdown();
+    //TODO try drop tables
+}
+
+void DatabaseAtomic::dropTableDataTask()
+{
+    LOG_INFO(log, String("Wake up ") + __PRETTY_FUNCTION__);
+    TableToDrop table;
+    try
+    {
+        std::lock_guard lock(tables_to_drop_mutex);
+        LOG_INFO(log, "There are " + std::to_string(tables_to_drop.size()) + " tables to drop");
+        time_t current_time = std::chrono::system_clock::to_time_t(std::chrono::system_clock::now());
+        auto it = std::find_if(tables_to_drop.begin(), tables_to_drop.end(), [current_time, this](const TableToDrop & elem)
+        {
+            LOG_INFO(log, "Check table " + elem.table->getStorageID().getNameForLogs() + ": " +
+                      "refcount = " + std::to_string(elem.table.unique()) + ", " +
+                      "time elapsed = " + std::to_string(current_time - elem.drop_time));
+            return elem.table.unique() && elem.drop_time + drop_delay_s < current_time;
+        });
+        if (it != tables_to_drop.end())
+        {
+            table = std::move(*it);
+            LOG_INFO(log, "Will try drop " + table.table->getStorageID().getNameForLogs());
+            tables_to_drop.erase(it);
+        }
+    }
+    catch (...)
+    {
+        tryLogCurrentException(log, __PRETTY_FUNCTION__);
+    }
+
+    if (table.table)
+    {
+        try
+        {
+            LOG_INFO(log, "Trying to drop table " + table.table->getStorageID().getNameForLogs());
+            table.table->drop();
+            table.table->is_dropped = true;
+            Poco::File table_data_dir{table.data_path};
+            if (table_data_dir.exists())
+                table_data_dir.remove(true);
+
+            String metadata_tmp_drop = getObjectMetadataPath(table.table->getStorageID().getTableName()) + drop_suffix;
+            Poco::File(metadata_tmp_drop).remove();
+        }
+        catch (...)
+        {
+            tryLogCurrentException(log, "Cannot drop table " + table.table->getStorageID().getNameForLogs() +
+                                        ". Will retry later.");
+            {
+                std::lock_guard lock(tables_to_drop_mutex);
+                tables_to_drop.emplace_back(std::move(table));
+            }
+        }
+    }
+
+    drop_task->scheduleAfter(reschedule_time_ms);
+}
+
 
 }
 
diff --git a/dbms/src/Databases/DatabaseAtomic.h b/dbms/src/Databases/DatabaseAtomic.h
index 42357e3b0ee408b1af77b131a84a03d71cd453a1..6e80df57eaa10bdf94cd161ea3508cdb6a574822 100644
--- a/dbms/src/Databases/DatabaseAtomic.h
+++ b/dbms/src/Databases/DatabaseAtomic.h
@@ -1,6 +1,7 @@
 #pragma once
 
 #include <Databases/DatabasesCommon.h>
+#include <Core/BackgroundSchedulePool.h>
 
 #include <Databases/DatabaseOrdinary.h>
 
@@ -11,7 +12,7 @@ class DatabaseAtomic : public DatabaseOrdinary
 {
 public:
 
-    DatabaseAtomic(String name_, String metadata_path_, const Context & context_);
+    DatabaseAtomic(String name_, String metadata_path_, Context & context_);
 
     String getEngineName() const override { return "Atomic"; }
 
@@ -21,6 +22,9 @@ public:
             IDatabase & to_database,
             const String & to_table_name) override;
 
+    //void removeTable(const Context & context, const String & table_name) override;
+    void dropTable(const Context & context, const String & table_name) override;
+
     void attachTable(const String & name, const StoragePtr & table, const String & relative_table_path = {}) override;
     StoragePtr detachTable(const String & name) override;
 
@@ -29,9 +33,30 @@ public:
 
     void drop(const Context & /*context*/) override;
 
+    void loadStoredObjects(Context & context, bool has_force_restore_data_flag) override;
+    void shutdown() override;
+
 private:
+    void dropTableDataTask();
+
+private:
+    static constexpr time_t drop_delay_s = 10;
+    static constexpr size_t reschedule_time_ms = 5000;
+
     //TODO store path in DatabaseWithOwnTables::tables
     std::map<String, String> table_name_to_path;
+    struct TableToDrop
+    {
+        StoragePtr table;
+        String data_path;
+        time_t drop_time;
+        //time_t last_attempt_time;
+    };
+    using TablesToDrop = std::list<TableToDrop>;
+    TablesToDrop tables_to_drop;
+    std::mutex tables_to_drop_mutex;
+
+    BackgroundSchedulePoolTaskHolder drop_task;
 
 };
 
diff --git a/dbms/src/Databases/DatabaseLazy.cpp b/dbms/src/Databases/DatabaseLazy.cpp
index 4a931694af07bfc3aad4bfe90a8c5126f887aae1..933514b57e52fbef9fa2c4ec234d154328cd092a 100644
--- a/dbms/src/Databases/DatabaseLazy.cpp
+++ b/dbms/src/Databases/DatabaseLazy.cpp
@@ -65,12 +65,12 @@ void DatabaseLazy::createTable(
         it->second.metadata_modification_time = DatabaseOnDisk::getObjectMetadataModificationTime(table_name);
 }
 
-void DatabaseLazy::removeTable(
+void DatabaseLazy::dropTable(
     const Context & context,
     const String & table_name)
 {
     SCOPE_EXIT({ clearExpiredTables(); });
-    DatabaseOnDisk::removeTable(context, table_name);
+    DatabaseOnDisk::dropTable(context, table_name);
 }
 
 void DatabaseLazy::renameTable(
diff --git a/dbms/src/Databases/DatabaseLazy.h b/dbms/src/Databases/DatabaseLazy.h
index 8fd04fea7c84fa2de840865da2da80f2bd03c32b..effda57a4ba146ca3b00cd8fb1f191a38e4bd545 100644
--- a/dbms/src/Databases/DatabaseLazy.h
+++ b/dbms/src/Databases/DatabaseLazy.h
@@ -32,7 +32,7 @@ public:
         const StoragePtr & table,
         const ASTPtr & query) override;
 
-    void removeTable(
+    void dropTable(
         const Context & context,
         const String & table_name) override;
 
diff --git a/dbms/src/Databases/DatabaseMemory.cpp b/dbms/src/Databases/DatabaseMemory.cpp
index 28a6a8a7af0602c8d965d41dbc63d0588c18757d..765d80c9feea5b25cebca558db8b07bbc871cae8 100644
--- a/dbms/src/Databases/DatabaseMemory.cpp
+++ b/dbms/src/Databases/DatabaseMemory.cpp
@@ -2,6 +2,8 @@
 #include <Databases/DatabaseMemory.h>
 #include <Databases/DatabasesCommon.h>
 #include <Parsers/ASTCreateQuery.h>
+#include <Storages/IStorage.h>
+#include <Poco/File.h>
 
 
 namespace DB
@@ -21,11 +23,23 @@ void DatabaseMemory::createTable(
     attachTable(table_name, table);
 }
 
-void DatabaseMemory::removeTable(
+void DatabaseMemory::dropTable(
     const Context & /*context*/,
     const String & table_name)
 {
-    detachTable(table_name);
+    auto table = detachTable(table_name);
+    try
+    {
+        table->drop();
+        Poco::File table_data_dir{getTableDataPath(table_name)};
+        if (table_data_dir.exists())
+            table_data_dir.remove(true);
+    }
+    catch (...)
+    {
+        attachTable(table_name, table);
+    }
+    table->is_dropped = true;
 }
 
 ASTPtr DatabaseMemory::getCreateDatabaseQuery(const Context & /*context*/) const
diff --git a/dbms/src/Databases/DatabaseMemory.h b/dbms/src/Databases/DatabaseMemory.h
index 01789c11901771d6d656cfba41c8f1abac8e8db1..27abf6e87be7b75d56b8d2252aff576d9431a6b8 100644
--- a/dbms/src/Databases/DatabaseMemory.h
+++ b/dbms/src/Databases/DatabaseMemory.h
@@ -29,7 +29,7 @@ public:
         const StoragePtr & table,
         const ASTPtr & query) override;
 
-    void removeTable(
+    void dropTable(
         const Context & context,
         const String & table_name) override;
 
diff --git a/dbms/src/Databases/DatabaseMySQL.cpp b/dbms/src/Databases/DatabaseMySQL.cpp
index 504909c6f1b989f61642deb5dd56d530594aba6f..4295afcb8a90eb4fb6ff482d714d0bf15aba9e7d 100644
--- a/dbms/src/Databases/DatabaseMySQL.cpp
+++ b/dbms/src/Databases/DatabaseMySQL.cpp
@@ -431,7 +431,7 @@ void DatabaseMySQL::loadStoredObjects(Context &, bool)
     }
 }
 
-void DatabaseMySQL::removeTable(const Context &, const String & table_name)
+void DatabaseMySQL::dropTable(const Context &, const String & table_name)
 {
     std::lock_guard<std::mutex> lock{mutex};
 
@@ -445,7 +445,8 @@ void DatabaseMySQL::removeTable(const Context &, const String & table_name)
         throw Exception("The remove flag file already exists but the " + backQuoteIfNeed(getDatabaseName()) +
             "." + backQuoteIfNeed(table_name) + " does not exists remove tables, it is bug.", ErrorCodes::LOGICAL_ERROR);
 
-    if (!local_tables_cache.count(table_name))
+    auto table_iter = local_tables_cache.find(table_name);
+    if (table_iter == local_tables_cache.end())
         throw Exception("Table " + backQuoteIfNeed(getDatabaseName()) + "." + backQuoteIfNeed(table_name) + " doesn't exist.",
             ErrorCodes::UNKNOWN_TABLE);
 
@@ -453,6 +454,7 @@ void DatabaseMySQL::removeTable(const Context &, const String & table_name)
 
     try
     {
+        table_iter->second.second->drop();
         remove_flag.createFile();
     }
     catch (...)
@@ -460,6 +462,7 @@ void DatabaseMySQL::removeTable(const Context &, const String & table_name)
         remove_or_detach_tables.erase(table_name);
         throw;
     }
+    table_iter->second.second->is_dropped = true;
 }
 
 DatabaseMySQL::~DatabaseMySQL()
diff --git a/dbms/src/Databases/DatabaseMySQL.h b/dbms/src/Databases/DatabaseMySQL.h
index fe4aa6d55aea10da2d3a8334697ff20232a57fdd..01885be8d70b35cd50f4a8cb348cd50869aaff99 100644
--- a/dbms/src/Databases/DatabaseMySQL.h
+++ b/dbms/src/Databases/DatabaseMySQL.h
@@ -1,6 +1,7 @@
 #pragma once
 
 #include "config_core.h"
+#define USE_MYSQL 1
 #if USE_MYSQL
 
 #include <mysqlxx/Pool.h>
@@ -52,7 +53,7 @@ public:
 
     StoragePtr detachTable(const String & table_name) override;
 
-    void removeTable(const Context &, const String & table_name) override;
+    void dropTable(const Context &, const String & table_name) override;
 
     void attachTable(const String & table_name, const StoragePtr & storage, const String & relative_table_path = {}) override;
 
diff --git a/dbms/src/Databases/DatabaseOnDisk.cpp b/dbms/src/Databases/DatabaseOnDisk.cpp
index 3d6ed5c25ffd24f1cb82a584625a4de9b001c05e..c9da90a3fd08a546c309a9ffcbd7a3c9261d6a39 100644
--- a/dbms/src/Databases/DatabaseOnDisk.cpp
+++ b/dbms/src/Databases/DatabaseOnDisk.cpp
@@ -159,7 +159,7 @@ void DatabaseOnDisk::createTable(
         throw Exception("Table " + backQuote(getDatabaseName()) + "." + backQuote(table_name) + " already exists.", ErrorCodes::TABLE_ALREADY_EXISTS);
 
     String table_metadata_path = getObjectMetadataPath(table_name);
-    String table_metadata_tmp_path = table_metadata_path + ".tmp";
+    String table_metadata_tmp_path = table_metadata_path + create_suffix;
     String statement;
 
     {
@@ -190,30 +190,36 @@ void DatabaseOnDisk::createTable(
     }
 }
 
-void DatabaseOnDisk::removeTable(const Context & /* context */, const String & table_name)
+void DatabaseOnDisk::dropTable(const Context &  context, const String & table_name)
 {
-    String table_data = getTableDataPath(table_name);
-    StoragePtr res = detachTable(table_name);
     String table_metadata_path = getObjectMetadataPath(table_name);
+    String table_metadata_path_drop = table_metadata_path + drop_suffix;
+    String table_data_path_relative = getTableDataPath(table_name);
+    assert(!table_data_path_relative.empty());
 
+    StoragePtr table = detachTable(table_name);
+    bool renamed = false;
     try
     {
-        Poco::File(table_metadata_path).remove();
+        Poco::File(table_metadata_path).renameTo(table_metadata_path_drop);
+        renamed = true;
+        table->drop();
+        table->is_dropped = true;
+
+        Poco::File table_data_dir{context.getPath() + table_data_path_relative};
+        if (table_data_dir.exists())
+            table_data_dir.remove(true);
     }
     catch (...)
     {
-        try
-        {
-            Poco::File(table_metadata_path + ".tmp_drop").remove();
-            return;
-        }
-        catch (...)
-        {
-            LOG_WARNING(log, getCurrentExceptionMessage(__PRETTY_FUNCTION__));
-        }
-        attachTable(table_name, res, data_path);
+        LOG_WARNING(log, getCurrentExceptionMessage(__PRETTY_FUNCTION__));
+        attachTable(table_name, table, table_data_path_relative);
+        if (renamed)
+            Poco::File(table_metadata_path_drop).renameTo(table_metadata_path);
         throw;
     }
+
+    Poco::File(table_metadata_path_drop).remove();
 }
 
 void DatabaseOnDisk::renameTable(
@@ -234,41 +240,43 @@ void DatabaseOnDisk::renameTable(
             throw Exception("Moving tables between databases of different engines is not supported", ErrorCodes::NOT_IMPLEMENTED);
     }
 
-    StoragePtr table = tryGetTable(context, table_name);
-
-    if (!table)
-        throw Exception("Table " + backQuote(getDatabaseName()) + "." + backQuote(table_name) + " doesn't exist.", ErrorCodes::UNKNOWN_TABLE);
-
-    auto table_lock = table->lockExclusively(context.getCurrentQueryId());
-
-    ASTPtr ast = parseQueryFromMetadata(context, getObjectMetadataPath(table_name));
-    auto & create = ast->as<ASTCreateQuery &>();
-    create.table = to_table_name;
-    if (from_ordinary_to_atomic)
-        create.uuid = UUIDHelpers::generateV4();
-    if (from_atomic_to_ordinary)
-        create.uuid = UUIDHelpers::Nil;
-
-    /// Notify the table that it is renamed. If the table does not support renaming, exception is thrown.
+    auto table_data_relative_path = getTableDataPath(table_name);
+    TableStructureWriteLockHolder table_lock;
+    String table_metadata_path;
+    ASTPtr attach_query;
+    StoragePtr table = detachTable(table_name);
     try
     {
-        table->rename(to_database.getTableDataPath(create),
-                      to_database.getDatabaseName(),
-                      to_table_name, table_lock);
+        table_lock = table->lockExclusively(context.getCurrentQueryId());
+
+        table_metadata_path = getObjectMetadataPath(table_name);
+        attach_query = parseQueryFromMetadata(context, table_metadata_path);
+        auto & create = attach_query->as<ASTCreateQuery &>();
+        create.table = to_table_name;
+        if (from_ordinary_to_atomic)
+            create.uuid = UUIDHelpers::generateV4();
+        if (from_atomic_to_ordinary)
+            create.uuid = UUIDHelpers::Nil;
+
+        /// Notify the table that it is renamed. It will move data to new path (if it stores data on disk) and update StorageID
+        table->rename(to_database.getTableDataPath(create), to_database.getDatabaseName(), to_table_name, table_lock);
     }
     catch (const Exception &)
     {
+        attachTable(table_name, table, table_data_relative_path);
         throw;
     }
     catch (const Poco::Exception & e)
     {
+        attachTable(table_name, table, table_data_relative_path);
         /// Better diagnostics.
         throw Exception{Exception::CreateFromPoco, e};
     }
 
-    /// NOTE Non-atomic.
-    to_database.createTable(context, to_table_name, table, ast);
-    removeTable(context, table_name);
+    /// Now table data are moved to new database, so we must add metadata and attach table to new database
+    to_database.createTable(context, to_table_name, table, attach_query);
+
+    Poco::File(table_metadata_path).remove();
 }
 
 ASTPtr DatabaseOnDisk::getCreateTableQueryImpl(const Context & context, const String & table_name, bool throw_on_error) const
diff --git a/dbms/src/Databases/DatabaseOnDisk.h b/dbms/src/Databases/DatabaseOnDisk.h
index 4025fabaef092f0781ca2dbd8679f8ca59b58738..602a435c6333d5af0072aaad4a02e9e41bc5c903 100644
--- a/dbms/src/Databases/DatabaseOnDisk.h
+++ b/dbms/src/Databases/DatabaseOnDisk.h
@@ -38,7 +38,7 @@ public:
         const StoragePtr & table,
         const ASTPtr & query) override;
 
-    void removeTable(
+    void dropTable(
         const Context & context,
         const String & table_name) override;
 
@@ -62,6 +62,9 @@ public:
     String getMetadataPath() const override { return metadata_path; }
 
 protected:
+    static constexpr const char * create_suffix = ".tmp";
+    static constexpr const char * drop_suffix = ".tmp_drop";
+
     using IteratingFunction = std::function<void(const String &)>;
     void iterateMetadataFiles(const Context & context, const IteratingFunction & iterating_function) const;
 
@@ -73,6 +76,8 @@ protected:
     ASTPtr parseQueryFromMetadata(const Context & context, const String & metadata_file_path, bool throw_on_error = true, bool remove_empty = false) const;
     ASTPtr getCreateQueryFromMetadata(const Context & context, const String & metadata_path, bool throw_on_error) const;
 
+    //bool detachTableAndRemoveMetadata(const String & table_name);
+    //void replaceMetadata(const ASTPtr & create, );
 
     const String metadata_path;
     /*const*/ String data_path;
diff --git a/dbms/src/Databases/IDatabase.h b/dbms/src/Databases/IDatabase.h
index 159bf45ce619f0ca41977734f14e63c882452509..de320cdba7ce90a4a8c3ed19c9d63de3658f404c 100644
--- a/dbms/src/Databases/IDatabase.h
+++ b/dbms/src/Databases/IDatabase.h
@@ -170,8 +170,8 @@ public:
         throw Exception("There is no CREATE DICTIONARY query for Database" + getEngineName(), ErrorCodes::NOT_IMPLEMENTED);
     }
 
-    /// Delete the table from the database. Delete the metadata.
-    virtual void removeTable(
+    /// Delete the table from the database, drop table and delete the metadata.
+    virtual void dropTable(
         const Context & /*context*/,
         const String & /*name*/)
     {
diff --git a/dbms/src/Formats/ParsedTemplateFormatString.cpp b/dbms/src/Formats/ParsedTemplateFormatString.cpp
index 981d43089a270598c3791a31b48e10b6650ff842..fe5810d8a0b70ff99522a878bbd0bb71a8bd27d4 100644
--- a/dbms/src/Formats/ParsedTemplateFormatString.cpp
+++ b/dbms/src/Formats/ParsedTemplateFormatString.cpp
@@ -16,11 +16,11 @@ namespace ErrorCodes
 
 ParsedTemplateFormatString::ParsedTemplateFormatString(const FormatSchemaInfo & schema, const ColumnIdxGetter & idx_by_name)
 {
+    ReadBufferFromFile schema_file(schema.absoluteSchemaPath(), 4096);
+    String format_string;
+    readStringUntilEOF(format_string, schema_file);
     try
     {
-        ReadBufferFromFile schema_file(schema.absoluteSchemaPath(), 4096);
-        String format_string;
-        readStringUntilEOF(format_string, schema_file);
         parse(format_string, idx_by_name);
     }
     catch (DB::Exception & e)
@@ -76,7 +76,14 @@ void ParsedTemplateFormatString::parse(const String & format_string, const Colum
 
             case Column:
                 column_names.emplace_back();
-                pos = readMayBeQuotedColumnNameInto(pos, end - pos, column_names.back());
+                try
+                {
+                    pos = readMayBeQuotedColumnNameInto(pos, end - pos, column_names.back());
+                }
+                catch (const DB::Exception & e)
+                {
+                    throwInvalidFormat(e.message(), columnsCount());
+                }
 
                 if (*pos == ':')
                     state = Format;
@@ -99,7 +106,16 @@ void ParsedTemplateFormatString::parse(const String & format_string, const Colum
                     errno = 0;
                     column_idx = strtoull(column_names.back().c_str(), &col_idx_end, 10);
                     if (col_idx_end != column_names.back().c_str() + column_names.back().size() || errno)
-                        column_idx = idx_by_name(column_names.back());
+                    {
+                        try
+                        {
+                            column_idx = idx_by_name(column_names.back());
+                        }
+                        catch (const DB::Exception & e)
+                        {
+                            throwInvalidFormat(e.message(), columnsCount());
+                        }
+                    }
                 }
                 format_idx_to_column_idx.emplace_back(column_idx);
                 break;
diff --git a/dbms/src/Interpreters/InterpreterDropQuery.cpp b/dbms/src/Interpreters/InterpreterDropQuery.cpp
index b1995869baf12ce6502d5d9e56075e3fdf4de30e..ad04dd4f41cce56578c6dbb9750b621f3d86b9ee 100644
--- a/dbms/src/Interpreters/InterpreterDropQuery.cpp
+++ b/dbms/src/Interpreters/InterpreterDropQuery.cpp
@@ -10,6 +10,7 @@
 #include <Common/escapeForFileName.h>
 #include <Common/quoteString.h>
 #include <Common/typeid_cast.h>
+#include <Databases/DatabaseAtomic.h>
 
 
 namespace DB
@@ -106,43 +107,7 @@ BlockIO InterpreterDropQuery::executeToTable(
 
             auto table_lock = database_and_table.second->lockExclusively(context.getCurrentQueryId());
 
-            const std::string metadata_file_without_extension =
-                database_and_table.first->getMetadataPath()
-                + escapeForFileName(table_id.table_name);
-
-            const auto prev_metadata_name = metadata_file_without_extension + ".sql";
-            const auto drop_metadata_name = metadata_file_without_extension + ".sql.tmp_drop";
-
-            /// Try to rename metadata file and delete the data
-            //TODO move this logic to DatabaseOnDisk
-            try
-            {
-                /// There some kind of tables that have no metadata - ignore renaming
-                if (Poco::File(prev_metadata_name).exists())
-                    Poco::File(prev_metadata_name).renameTo(drop_metadata_name);
-                /// Delete table data
-                database_and_table.second->drop(table_lock);
-            }
-            catch (...)
-            {
-                if (Poco::File(drop_metadata_name).exists())
-                    Poco::File(drop_metadata_name).renameTo(prev_metadata_name);
-                throw;
-            }
-
-            String table_data_path_relative = database_and_table.first->getTableDataPath(table_name);
-
-            /// Delete table metadata and table itself from memory
-            database_and_table.first->removeTable(context, table_id.table_name);
-            database_and_table.second->is_dropped = true;
-
-            /// If it is not virtual database like Dictionary then drop remaining data dir
-            if (!table_data_path_relative.empty())
-            {
-                String table_data_path = context.getPath() + table_data_path_relative;
-                if (Poco::File(table_data_path).exists())
-                    Poco::File(table_data_path).remove(true);
-            }
+            database_and_table.first->dropTable(context, table_name);
         }
     }
 
@@ -217,7 +182,7 @@ BlockIO InterpreterDropQuery::executeToTemporaryTable(String & table_name, ASTDr
                 /// If table was already dropped by anyone, an exception will be thrown
                 auto table_lock = table->lockExclusively(context.getCurrentQueryId());
                 /// Delete table data
-                table->drop(table_lock);
+                table->drop();
                 table->is_dropped = true;
             }
         }
diff --git a/dbms/src/Storages/IStorage.h b/dbms/src/Storages/IStorage.h
index dcb805b141e84db7d7e632c9fd27eb4bd1f526d6..a5655d65903fb721b1c70385e232b970bf5da14d 100644
--- a/dbms/src/Storages/IStorage.h
+++ b/dbms/src/Storages/IStorage.h
@@ -291,9 +291,10 @@ public:
     /** Delete the table data. Called before deleting the directory with the data.
       * The method can be called only after detaching table from Context (when no queries are performed with table).
       * The table is not usable during and after call to this method.
+      * If some queries may still use the table, then it must be called under exclusive lock.
       * If you do not need any action other than deleting the directory with data, you can leave this method blank.
       */
-    virtual void drop(TableStructureWriteLockHolder &) {}
+    virtual void drop() {}
 
     /** Clear the table data and leave it empty.
       * Must be called under lockForAlter.
diff --git a/dbms/src/Storages/LiveView/StorageLiveView.cpp b/dbms/src/Storages/LiveView/StorageLiveView.cpp
index 6eca05d2c483a730c19696a35c1789211bfb78ba..ca804b28b4a3faea7e4792157895e276346c7be3 100644
--- a/dbms/src/Storages/LiveView/StorageLiveView.cpp
+++ b/dbms/src/Storages/LiveView/StorageLiveView.cpp
@@ -487,7 +487,7 @@ StorageLiveView::~StorageLiveView()
     }
 }
 
-void StorageLiveView::drop(TableStructureWriteLockHolder &)
+void StorageLiveView::drop()
 {
     auto table_id = getStorageID();
     global_context.removeDependency(select_table_id, table_id);
diff --git a/dbms/src/Storages/LiveView/StorageLiveView.h b/dbms/src/Storages/LiveView/StorageLiveView.h
index 77992b97e1936f816a644b0f492adf727be95f89..ba8e56317c81616aef507d5798a3a6d519e530e5 100644
--- a/dbms/src/Storages/LiveView/StorageLiveView.h
+++ b/dbms/src/Storages/LiveView/StorageLiveView.h
@@ -124,7 +124,7 @@ public:
     }
 
     void checkTableCanBeDropped() const override;
-    void drop(TableStructureWriteLockHolder &) override;
+    void drop() override;
     void startup() override;
     void shutdown() override;
 
diff --git a/dbms/src/Storages/MergeTree/MergeTreeData.cpp b/dbms/src/Storages/MergeTree/MergeTreeData.cpp
index c47afbf407eb05abb1a3911d95fa5075ec41c813..a060813f35122e97afb55063dbc17b14bedd04f8 100644
--- a/dbms/src/Storages/MergeTree/MergeTreeData.cpp
+++ b/dbms/src/Storages/MergeTree/MergeTreeData.cpp
@@ -1381,7 +1381,8 @@ void MergeTreeData::dropAllData()
     auto full_paths = getDataPaths();
 
     for (auto && full_data_path : full_paths)
-        Poco::File(full_data_path).remove(true);
+        if (Poco::File(full_data_path).exists())
+            Poco::File(full_data_path).remove(true);
 
     LOG_TRACE(log, "dropAllData: done.");
 }
diff --git a/dbms/src/Storages/StorageDistributed.h b/dbms/src/Storages/StorageDistributed.h
index c1e3c639c1fa166559c0ec565c14331fe5c580fe..7a3ecdc377b2dc51be2683b8004d924dde6a761e 100644
--- a/dbms/src/Storages/StorageDistributed.h
+++ b/dbms/src/Storages/StorageDistributed.h
@@ -75,8 +75,6 @@ public:
 
     BlockOutputStreamPtr write(const ASTPtr & query, const Context & context) override;
 
-    void drop(TableStructureWriteLockHolder &) override {}
-
     /// Removes temporary data in local filesystem.
     void truncate(const ASTPtr &, const Context &, TableStructureWriteLockHolder &) override;
 
diff --git a/dbms/src/Storages/StorageMaterializedView.cpp b/dbms/src/Storages/StorageMaterializedView.cpp
index 2bb79dcaa1d06c97abde479edf483add94ff3e96..10dd3704680404340f06c69e7106a0ca42fccb59 100644
--- a/dbms/src/Storages/StorageMaterializedView.cpp
+++ b/dbms/src/Storages/StorageMaterializedView.cpp
@@ -209,7 +209,7 @@ static void executeDropQuery(ASTDropQuery::Kind kind, Context & global_context,
 }
 
 
-void StorageMaterializedView::drop(TableStructureWriteLockHolder &)
+void StorageMaterializedView::drop()
 {
     auto table_id = getStorageID();
     if (!select_table_id.empty())
diff --git a/dbms/src/Storages/StorageMaterializedView.h b/dbms/src/Storages/StorageMaterializedView.h
index f9a57783871d05fb90735152d454764c99d56c77..3af066b961946cc5edd68f994c3ade974c95c4bd 100644
--- a/dbms/src/Storages/StorageMaterializedView.h
+++ b/dbms/src/Storages/StorageMaterializedView.h
@@ -31,7 +31,7 @@ public:
 
     BlockOutputStreamPtr write(const ASTPtr & query, const Context & context) override;
 
-    void drop(TableStructureWriteLockHolder &) override;
+    void drop() override;
 
     void truncate(const ASTPtr &, const Context &, TableStructureWriteLockHolder &) override;
 
diff --git a/dbms/src/Storages/StorageMemory.cpp b/dbms/src/Storages/StorageMemory.cpp
index b2cb5682f6d90dad382e429f74a3f16c4cc61fef..fbef2c753316426b62672a8c57f430c69c70096a 100644
--- a/dbms/src/Storages/StorageMemory.cpp
+++ b/dbms/src/Storages/StorageMemory.cpp
@@ -123,7 +123,7 @@ BlockOutputStreamPtr StorageMemory::write(
 }
 
 
-void StorageMemory::drop(TableStructureWriteLockHolder &)
+void StorageMemory::drop()
 {
     std::lock_guard lock(mutex);
     data.clear();
diff --git a/dbms/src/Storages/StorageMemory.h b/dbms/src/Storages/StorageMemory.h
index 1e66b17606b32172319132f611574a70ecaf95d3..fd21cf356ff19309caa170b3bb5e5aeb41a3a887 100644
--- a/dbms/src/Storages/StorageMemory.h
+++ b/dbms/src/Storages/StorageMemory.h
@@ -38,7 +38,7 @@ public:
 
     BlockOutputStreamPtr write(const ASTPtr & query, const Context & context) override;
 
-    void drop(TableStructureWriteLockHolder &) override;
+    void drop() override;
 
     void truncate(const ASTPtr &, const Context &, TableStructureWriteLockHolder &) override;
 
diff --git a/dbms/src/Storages/StorageMergeTree.cpp b/dbms/src/Storages/StorageMergeTree.cpp
index f5279ebef923f00aae4a206785c7f8e7317d1051..ccc04de7ee6b27e982d7c61fe8f63052cc1d60d3 100644
--- a/dbms/src/Storages/StorageMergeTree.cpp
+++ b/dbms/src/Storages/StorageMergeTree.cpp
@@ -170,7 +170,7 @@ void StorageMergeTree::checkPartitionCanBeDropped(const ASTPtr & partition)
     global_context.checkPartitionCanBeDropped(table_id.database_name, table_id.table_name, partition_size);
 }
 
-void StorageMergeTree::drop(TableStructureWriteLockHolder &)
+void StorageMergeTree::drop()
 {
     shutdown();
     dropAllData();
diff --git a/dbms/src/Storages/StorageMergeTree.h b/dbms/src/Storages/StorageMergeTree.h
index 765b43ed90ec3d4a5e27125cd942bf0a052b9b97..43d38c40de279fa94ab2a84066fb9148dbb77eb8 100644
--- a/dbms/src/Storages/StorageMergeTree.h
+++ b/dbms/src/Storages/StorageMergeTree.h
@@ -62,7 +62,7 @@ public:
 
     CancellationCode killMutation(const String & mutation_id) override;
 
-    void drop(TableStructureWriteLockHolder &) override;
+    void drop() override;
     void truncate(const ASTPtr &, const Context &, TableStructureWriteLockHolder &) override;
 
     void alter(const AlterCommands & commands, const Context & context, TableStructureWriteLockHolder & table_lock_holder) override;
diff --git a/dbms/src/Storages/StorageReplicatedMergeTree.cpp b/dbms/src/Storages/StorageReplicatedMergeTree.cpp
index 2fdd7daa684490caa4a6bee57ab4f96b49775f26..96428573ddb4c216cc0b7c6735bdad01fd552f38 100644
--- a/dbms/src/Storages/StorageReplicatedMergeTree.cpp
+++ b/dbms/src/Storages/StorageReplicatedMergeTree.cpp
@@ -3798,7 +3798,7 @@ void StorageReplicatedMergeTree::checkPartitionCanBeDropped(const ASTPtr & parti
 }
 
 
-void StorageReplicatedMergeTree::drop(TableStructureWriteLockHolder &)
+void StorageReplicatedMergeTree::drop()
 {
     {
         auto zookeeper = tryGetZooKeeper();
diff --git a/dbms/src/Storages/StorageReplicatedMergeTree.h b/dbms/src/Storages/StorageReplicatedMergeTree.h
index 0fff99b00f3da51d6433f0e8b1f3c81e92a71810..bbe12a05071f2bb3184e424cdadd67d9510bc5b2 100644
--- a/dbms/src/Storages/StorageReplicatedMergeTree.h
+++ b/dbms/src/Storages/StorageReplicatedMergeTree.h
@@ -113,7 +113,7 @@ public:
 
     /** Removes a replica from ZooKeeper. If there are no other replicas, it deletes the entire table from ZooKeeper.
       */
-    void drop(TableStructureWriteLockHolder &) override;
+    void drop() override;
 
     void truncate(const ASTPtr &, const Context &, TableStructureWriteLockHolder &) override;
 
diff --git a/dbms/src/Storages/StorageTinyLog.cpp b/dbms/src/Storages/StorageTinyLog.cpp
index aeb90399816ba7c85f8c7df9d7fb4407dc55a3a5..2b9ac0df296758d0b24f6a74b238f632d25b703d 100644
--- a/dbms/src/Storages/StorageTinyLog.cpp
+++ b/dbms/src/Storages/StorageTinyLog.cpp
@@ -425,10 +425,13 @@ void StorageTinyLog::truncate(const ASTPtr &, const Context &, TableStructureWri
         addFiles(column.name, *column.type);
 }
 
-void StorageTinyLog::drop(TableStructureWriteLockHolder &)
+void StorageTinyLog::drop()
 {
+    // TODO do we really need another rwlock here if drop() is called under exclusive table lock?
     std::unique_lock<std::shared_mutex> lock(rwlock);
-    disk->removeRecursive(table_path);
+    // TODO may be move it to DatabaseOnDisk
+    if (disk->exists(table_path))
+        disk->removeRecursive(table_path);
     files.clear();
 }
 
diff --git a/dbms/src/Storages/StorageTinyLog.h b/dbms/src/Storages/StorageTinyLog.h
index b9a45a9f271c6352744af013cb89be5e84ea6927..ed75849d84357db8ba1c38da40103c5ce1ad4953 100644
--- a/dbms/src/Storages/StorageTinyLog.h
+++ b/dbms/src/Storages/StorageTinyLog.h
@@ -46,7 +46,7 @@ public:
 
     void truncate(const ASTPtr &, const Context &, TableStructureWriteLockHolder &) override;
 
-    void drop(TableStructureWriteLockHolder &) override;
+    void drop() override;
 
 protected:
     StorageTinyLog(