Revert "Revert "Merge pull request #45493 from azat/fix-detach""

This reverts commit 9dc4f02d15.

src/Databases/DatabaseOnDisk.cpp

@@ -379,14 +379,14 @@ void DatabaseOnDisk::renameTable(
     if (dictionary && table && !table->isDictionary())
         throw Exception(ErrorCodes::INCORRECT_QUERY, "Use RENAME/EXCHANGE TABLE (instead of RENAME/EXCHANGE DICTIONARY) for tables");
 
+    table_lock = table->lockExclusively(
+        local_context->getCurrentQueryId(), local_context->getSettingsRef().lock_acquire_timeout);
+
     detachTable(local_context, table_name);
 
     UUID prev_uuid = UUIDHelpers::Nil;
     try
     {
-        table_lock = table->lockExclusively(
-            local_context->getCurrentQueryId(), local_context->getSettingsRef().lock_acquire_timeout);
-
         table_metadata_path = getObjectMetadataPath(table_name);
         attach_query = parseQueryFromMetadata(log, local_context, table_metadata_path);
         auto & create = attach_query->as<ASTCreateQuery &>();

src/Databases/DatabasesCommon.cpp

@@ -236,6 +236,7 @@ StoragePtr DatabaseWithOwnTablesBase::detachTableUnlocked(const String & table_n
                         backQuote(database_name), backQuote(table_name));
     res = it->second;
     tables.erase(it);
+    res->is_detached = true;
 
     auto table_id = res->getStorageID();
     if (table_id.hasUUID())
@@ -272,6 +273,10 @@ void DatabaseWithOwnTablesBase::attachTableUnlocked(const String & table_name, c
             DatabaseCatalog::instance().removeUUIDMapping(table_id.uuid);
         throw Exception(ErrorCodes::TABLE_ALREADY_EXISTS, "Table {} already exists.", table_id.getFullTableName());
     }
+
+    /// It is important to reset is_detached here since in case of RENAME in
+    /// non-Atomic database the is_detached is set to true before RENAME.
+    table->is_detached = false;
 }
 
 void DatabaseWithOwnTablesBase::shutdown()

src/Databases/MySQL/DatabaseMySQL.cpp

@@ -448,7 +448,7 @@ void DatabaseMySQL::detachTablePermanently(ContextPtr, const String & table_name
         remove_or_detach_tables.erase(table_name);
         throw;
     }
-    table_iter->second.second->is_dropped = true;
+    table_iter->second.second->is_detached = true;
 }
 
 void DatabaseMySQL::dropTable(ContextPtr local_context, const String & table_name, bool /*sync*/)

src/Interpreters/InterpreterDropQuery.cpp

@@ -287,6 +287,10 @@ BlockIO InterpreterDropQuery::executeToTemporaryTable(const String & table_name,
                 table->drop();
                 table->is_dropped = true;
             }
+            else if (kind == ASTDropQuery::Kind::Detach)
+            {
+                table->is_detached = true;
+            }
         }
     }
 

src/Storages/IStorage.cpp

@@ -50,10 +50,10 @@ TableLockHolder IStorage::lockForShare(const String & query_id, const std::chron
 {
     TableLockHolder result = tryLockTimed(drop_lock, RWLockImpl::Read, query_id, acquire_timeout);
 
-    if (is_dropped)
+    if (is_dropped || is_detached)
     {
         auto table_id = getStorageID();
-        throw Exception(ErrorCodes::TABLE_IS_DROPPED, "Table {}.{} is dropped", table_id.database_name, table_id.table_name);
+        throw Exception(ErrorCodes::TABLE_IS_DROPPED, "Table {}.{} is dropped or detached", table_id.database_name, table_id.table_name);
     }
     return result;
 }
@@ -62,7 +62,7 @@ TableLockHolder IStorage::tryLockForShare(const String & query_id, const std::ch
 {
     TableLockHolder result = tryLockTimed(drop_lock, RWLockImpl::Read, query_id, acquire_timeout);
 
-    if (is_dropped)
+    if (is_dropped || is_detached)
     {
         // Table was dropped while acquiring the lock
         result = nullptr;
@@ -81,7 +81,7 @@ IStorage::AlterLockHolder IStorage::lockForAlter(const std::chrono::milliseconds
                         "Possible deadlock avoided. Client should retry.",
                         getStorageID().getFullTableName(), acquire_timeout.count());
 
-    if (is_dropped)
+    if (is_dropped || is_detached)
         throw Exception(ErrorCodes::TABLE_IS_DROPPED, "Table {} is dropped or detached", getStorageID());
 
     return lock;
@@ -93,7 +93,7 @@ TableExclusiveLockHolder IStorage::lockExclusively(const String & query_id, cons
     TableExclusiveLockHolder result;
     result.drop_lock = tryLockTimed(drop_lock, RWLockImpl::Write, query_id, acquire_timeout);
 
-    if (is_dropped)
+    if (is_dropped || is_detached)
         throw Exception(ErrorCodes::TABLE_IS_DROPPED, "Table {} is dropped or detached", getStorageID());
 
     return result;

src/Storages/IStorage.h

@@ -562,6 +562,7 @@ public:
     virtual void onActionLockRemove(StorageActionBlockType /* action_type */) {}
 
     std::atomic<bool> is_dropped{false};
+    std::atomic<bool> is_detached{false};
 
     /// Does table support index for IN sections
     virtual bool supportsIndexForIn() const { return false; }

src/Storages/LiveView/StorageLiveView.cpp

@@ -470,7 +470,6 @@ void StorageLiveView::drop()
     DatabaseCatalog::instance().removeViewDependency(select_table_id, table_id);
 
     std::lock_guard lock(mutex);
-    is_dropped = true;
     condition.notify_all();
 }