Merge pull request #6413 from nikitamikhaylov/atomic_zookeeper2

Atomicity of Replicated table removal.

dbms/src/Databases/DatabaseOrdinary.cpp

@@ -135,7 +135,25 @@ void DatabaseOrdinary::loadTables(
         if (endsWith(dir_it.name(), ".sql.bak"))
             continue;
 
-        /// There are files .sql.tmp - delete.
+        // There are files that we tried to delete previously
+        const std::string tmp_drop_ext = ".sql.tmp_drop";
+        if (endsWith(dir_it.name(), ".sql.tmp_drop"))
+        {
+            const std::string table_name = dir_it.name().substr(0, dir_it.name().size() - tmp_drop_ext.size());
+            if (Poco::File(data_path + '/' + table_name).exists())
+            {
+                Poco::File(dir_it->path()).renameTo(table_name + ".sql");
+                LOG_WARNING(log, "Table was not dropped previously");
+            }
+            else
+            {
+                LOG_INFO(log, "Removing file " << dir_it->path());
+                Poco::File(dir_it->path()).remove();
+            }
+            continue;
+        }
+
+        /// There are files .sql.tmp - delete
         if (endsWith(dir_it.name(), ".sql.tmp"))
         {
             LOG_INFO(log, "Removing file " << dir_it->path());
@@ -302,6 +320,12 @@ void DatabaseOrdinary::removeTable(
     }
     catch (...)
     {
+        try
+        {
+            Poco::File(table_metadata_path + ".tmp_drop").remove();
+            return;
+        }
+        catch (...) {}
         attachTable(table_name, res);
         throw;
     }

dbms/src/Interpreters/InterpreterDropQuery.cpp

@@ -90,11 +90,28 @@ BlockIO InterpreterDropQuery::executeToTable(String & database_name_, String & t
             /// If table was already dropped by anyone, an exception will be thrown
 
             auto table_lock = database_and_table.second->lockExclusively(context.getCurrentQueryId());
-            /// Delete table metadata and table itself from memory
 
+            const auto prev_metadata_name = database_and_table.first->getMetadataPath() + escapeForFileName(database_and_table.second->getTableName()) + ".sql";
+            const auto drop_metadata_name = database_and_table.first->getMetadataPath() + escapeForFileName(database_and_table.second->getTableName()) + ".sql.tmp_drop";
+
+            /// Try to rename metadata file and delete the data
+            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;
+            }
+
+            /// Delete table metadata and table itself from memory
             database_and_table.first->removeTable(context, database_and_table.second->getTableName());
-            /// Delete table data
-            database_and_table.second->drop(table_lock);
             database_and_table.second->is_dropped = true;
 
             String database_data_path = database_and_table.first->getDataPath();

dbms/tests/integration/test_atomic_drop_table/configs/config.d/zookeeper_session_timeout.xml

@@ -0,0 +1,6 @@
+<yandex>
+    <zookeeper>
+        <!-- Required for correct timing in current test case -->
+        <session_timeout_ms replace="1">3000</session_timeout_ms>
+    </zookeeper>
+</yandex>

dbms/tests/integration/test_atomic_drop_table/configs/remote_servers.xml

@@ -0,0 +1,14 @@
+<yandex>
+    <remote_servers>
+        <test_cluster>
+            <shard>
+                <internal_replication>true</internal_replication>
+                <replica>
+                    <default_database>shard_0</default_database>
+                    <host>node1</host>
+                    <port>9000</port>
+                </replica>
+            </shard>
+        </test_cluster>
+    </remote_servers>
+</yandex>

dbms/tests/integration/test_atomic_drop_table/test.py

@@ -0,0 +1,37 @@
+import time
+import pytest
+
+from helpers.network import PartitionManager
+from helpers.cluster import ClickHouseCluster
+
+
+cluster = ClickHouseCluster(__file__)
+node1 = cluster.add_instance('node1', config_dir="configs", with_zookeeper=True)
+
+
+@pytest.fixture(scope="module")
+def start_cluster():
+    try:
+        cluster.start()
+        node1.query("CREATE DATABASE zktest;")
+        node1.query(
+            '''
+            CREATE TABLE zktest.atomic_drop_table (n UInt32)
+            ENGINE = ReplicatedMergeTree('/clickhouse/zktest/tables/atomic_drop_table', 'node1')
+            PARTITION BY n ORDER BY n
+            '''
+        )
+        yield cluster
+    finally:
+        cluster.shutdown()
+
+def test_atomic_delete_with_stopped_zookeeper(start_cluster):
+    node1.query("insert into zktest.atomic_drop_table values (8192)")
+
+    with PartitionManager() as pm:
+        pm.drop_instance_zk_connections(node1)
+        error = node1.query_and_get_error("DROP TABLE zktest.atomic_drop_table") #Table won't drop
+        assert error != ""
+
+    time.sleep(5)
+    assert '8192' in node1.query("select * from zktest.atomic_drop_table")