This commit is contained in:
Mikhail Filimonov 2020-12-15 18:06:23 +01:00
parent 27ff3a5214
commit 9df7ecb8e6
No known key found for this signature in database
GPG Key ID: 6E49C2E9AF1220BE
7 changed files with 78 additions and 21 deletions

View File

@ -169,17 +169,33 @@ void DatabaseOnDisk::createTable(
if (isTableExist(table_name, global_context))
throw Exception(ErrorCodes::TABLE_ALREADY_EXISTS, "Table {}.{} already exists", backQuote(getDatabaseName()), backQuote(table_name));
if (!create.attach)
checkMetadataFilenameAvailability(table_name);
String table_metadata_path = getObjectMetadataPath(table_name);
if (create.attach_short_syntax)
{
/// Metadata already exists, table was detached
attachTable(table_name, table, getTableDataPath(create));
removeDetachedPermanentlyFlag(table_name, table_metadata_path);
return;
}
String table_metadata_path = getObjectMetadataPath(table_name);
if (!create.attach)
checkMetadataFilenameAvailability(table_name);
if (create.attach && Poco::File(table_metadata_path).exists())
{
ASTPtr ast_detached = parseQueryFromMetadata(log, context, table_metadata_path);
auto & create_detached = ast_detached->as<ASTCreateQuery &>();
// either both should be Nil, either values should be equal
if (create.uuid != create_detached.uuid)
throw Exception(
ErrorCodes::TABLE_ALREADY_EXISTS,
"Table {}.{} already exist (detached permanently). To attach it back "
"you need to use short ATTACH syntax or a full statement with the same UUID",
backQuote(getDatabaseName()), backQuote(table_name));
}
String table_metadata_tmp_path = table_metadata_path + create_suffix;
String statement;
@ -196,6 +212,26 @@ void DatabaseOnDisk::createTable(
}
commitCreateTable(create, table, table_metadata_tmp_path, table_metadata_path);
removeDetachedPermanentlyFlag(table_name, table_metadata_path);
}
/// If the table was detached permanently we will have a flag file with
/// .sql.detached extension, is not needed anymore since we attached the table back
void DatabaseOnDisk::removeDetachedPermanentlyFlag(const String & table_name, const String & table_metadata_path) const
{
try
{
auto detached_permanently_flag = Poco::File(table_metadata_path + detached_suffix);
if (detached_permanently_flag.exists())
detached_permanently_flag.remove();
}
catch (Exception & e)
{
e.addMessage("while trying to remove permanenty detached flag. Table {}.{} may still be marked as permanently detached, and will not be reattached during server restart.", backQuote(getDatabaseName()), backQuote(table_name));
throw;
}
}
void DatabaseOnDisk::commitCreateTable(const ASTCreateQuery & query, const StoragePtr & table,
@ -215,20 +251,6 @@ void DatabaseOnDisk::commitCreateTable(const ASTCreateQuery & query, const Stora
Poco::File(table_metadata_tmp_path).remove();
throw;
}
try
{
/// If the table was detached permanently we will have a flag file with
/// .sql.detached extension, which is not needed anymore since we attached the table back
auto detached_permanently_flag = Poco::File(table_metadata_path + detached_suffix);
if (detached_permanently_flag.exists())
detached_permanently_flag.remove();
}
catch (Exception & e)
{
e.addMessage("while trying to remove permanenty detached flag. Table {}.{} may still be marked as permanently detached, and will not be reattached during server restart.", backQuote(getDatabaseName()), backQuote(query.table));
throw;
}
}
void DatabaseOnDisk::detachTablePermanently(const String & table_name)
@ -288,7 +310,7 @@ void DatabaseOnDisk::dropTable(const Context & context, const String & table_nam
void DatabaseOnDisk::checkMetadataFilenameAvailability(const String & to_table_name) const
{
std::unique_lock lock(mutex);
checkMetadataFilenameAvailabilityUnlocked(create, lock);
checkMetadataFilenameAvailabilityUnlocked(to_table_name, lock);
}
void DatabaseOnDisk::checkMetadataFilenameAvailabilityUnlocked(const String & to_table_name, std::unique_lock<std::mutex> &) const

View File

@ -94,6 +94,9 @@ protected:
const String metadata_path;
const String data_path;
private:
void removeDetachedPermanentlyFlag(const String & table_name, const String & table_metadata_path) const;
};
}

View File

@ -139,9 +139,9 @@ void DatabaseOrdinary::loadStoredObjects(Context & context, bool has_force_resto
auto detached_permanently_flag = Poco::File(full_path.string() + detached_suffix);
if (detached_permanently_flag.exists())
{
/// even if we don't load the table we still mark the uuid of it as taken.
if (create_query->uuid != UUIDHelpers::Nil)
DatabaseCatalog::instance().addUUIDMapping(create_query->uuid);
/// FIXME: even if we don't load the table we can still mark the uuid of it as taken.
/// if (create_query->uuid != UUIDHelpers::Nil)
/// DatabaseCatalog::instance().addUUIDMapping(create_query->uuid);
const std::string table_name = file_name.substr(0, file_name.size() - 4);
LOG_DEBUG(log, "Skipping permanently detached table {}.", backQuote(table_name));

View File

@ -15,7 +15,9 @@ ORDER BY tuple()
SETTINGS index_granularity = 8192
can not attach with bad uuid
can attach with short syntax
100
can not detach permanently the table which is already detached (temporary)
100
After database reattachement the table is back (it was detached temporary)
And we can detach it permanently
After database reattachement the table is still absent (it was detached permamently)

View File

@ -34,6 +34,8 @@ ATTACH TABLE test1601_detach_permanently_atomic.test_name_reuse UUID '00000000-0
SELECT 'can attach with short syntax';
ATTACH TABLE test1601_detach_permanently_atomic.test_name_reuse;
SELECT count() FROM test1601_detach_permanently_atomic.test_name_reuse;
DETACH table test1601_detach_permanently_atomic.test_name_reuse;
SELECT 'can not detach permanently the table which is already detached (temporary)';
@ -42,6 +44,8 @@ DETACH table test1601_detach_permanently_atomic.test_name_reuse PERMANENTLY; --
DETACH DATABASE test1601_detach_permanently_atomic;
ATTACH DATABASE test1601_detach_permanently_atomic;
SELECT count() FROM test1601_detach_permanently_atomic.test_name_reuse;
SELECT 'After database reattachement the table is back (it was detached temporary)';
SELECT 'And we can detach it permanently';
DETACH table test1601_detach_permanently_atomic.test_name_reuse PERMANENTLY;

View File

@ -0,0 +1,4 @@
0
200
0
200

View File

@ -0,0 +1,22 @@
DROP database IF EXISTS test_1603_rename_bug_ordinary;
create database test_1603_rename_bug_ordinary engine=Ordinary;
create table test_1603_rename_bug_ordinary.foo engine=Memory as select * from numbers(100);
create table test_1603_rename_bug_ordinary.bar engine=Log as select * from numbers(200);
detach table test_1603_rename_bug_ordinary.foo;
rename table test_1603_rename_bug_ordinary.bar to test_1603_rename_bug_ordinary.foo; -- { serverError 57 }
attach table test_1603_rename_bug_ordinary.foo;
SELECT count() from test_1603_rename_bug_ordinary.foo;
SELECT count() from test_1603_rename_bug_ordinary.bar;
DROP DATABASE test_1603_rename_bug_ordinary;
-- was not broken, adding just in case.
DROP database IF EXISTS test_1603_rename_bug_atomic;
create database test_1603_rename_bug_atomic engine=Atomic;
create table test_1603_rename_bug_atomic.foo engine=Memory as select * from numbers(100);
create table test_1603_rename_bug_atomic.bar engine=Log as select * from numbers(200);
detach table test_1603_rename_bug_atomic.foo;
rename table test_1603_rename_bug_atomic.bar to test_1603_rename_bug_atomic.foo; -- { serverError 57 }
attach table test_1603_rename_bug_atomic.foo;
SELECT count() from test_1603_rename_bug_atomic.foo;
SELECT count() from test_1603_rename_bug_atomic.bar;
DROP DATABASE test_1603_rename_bug_atomic;