thevar1able/ClickHouse
1
0
Fork 0
mirror of https://github.com/ClickHouse/ClickHouse.git synced 2024-12-17 11:52:27 +00:00
Code Issues Packages Projects Releases Wiki Activity

Better

Browse Source
This commit is contained in:
kssenii 2021-04-22 08:21:37 +00:00
parent 7d1e5e6a08
commit 814e727d0e
5 changed files with 121 additions and 121 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

105
src/Disks/HDFS/DiskHDFS.cpp
View File

@ -5,7 +5,7 @@
#include "ReadIndirectBufferFromHDFS.h"
#include "WriteIndirectBufferFromHDFS.h"
#include <IO/SeekAvoidingReadBuffer.h>
#include <Common/checkStackSize.h>
#include <Common/quoteString.h>
#include <common/logger_useful.h>
@ -18,6 +18,8 @@ namespace ErrorCodes
extern const int BAD_ARGUMENTS;
extern const int LOGICAL_ERROR;
extern const int PATH_ACCESS_DENIED;
extern const int CANNOT_DELETE_DIRECTORY;
extern const int UNKNOWN_FORMAT;
}
@ -104,6 +106,106 @@ void DiskHDFS::removeFromRemoteFS(const Metadata & metadata)
}
void DiskHDFS::removeSharedFile(const String & path, bool keep_in_remote_fs)
{
removeMeta(path, keep_in_remote_fs);
}
void DiskHDFS::removeSharedRecursive(const String & path, bool keep_in_remote_fs)
{
removeMetaRecursive(path, keep_in_remote_fs);
}
void DiskHDFS::removeFileIfExists(const String & path)
{
if (Poco::File(metadata_path + path).exists())
removeMeta(path, /* keep_in_remote_fs */ false);
}
void DiskHDFS::removeRecursive(const String & path)
{
checkStackSize(); /// This is needed to prevent stack overflow in case of cyclic symlinks.
Poco::File file(metadata_path + path);
if (file.isFile())
{
removeFile(path);
}
else
{
for (auto it{iterateDirectory(path)}; it->isValid(); it->next())
removeRecursive(it->path());
file.remove();
}
}
void DiskHDFS::removeMeta(const String & path, bool keep_in_remote_fs)
{
Poco::File file(metadata_path + path);
if (!file.isFile())
throw Exception(ErrorCodes::CANNOT_DELETE_DIRECTORY, "Path '{}' is a directory", path);
try
{
auto metadata = readMeta(path);
/// If there is no references - delete content from remote FS.
if (metadata.ref_count == 0)
{
file.remove();
if (!keep_in_remote_fs)
removeFromRemoteFS(metadata);
}
else /// In other case decrement number of references, save metadata and delete file.
{
--metadata.ref_count;
metadata.save();
file.remove();
}
}
catch (const Exception & e)
{
/// If it's impossible to read meta - just remove it from FS.
if (e.code() == ErrorCodes::UNKNOWN_FORMAT)
{
LOG_WARNING(
log,
"Metadata file {} can't be read by reason: {}. Removing it forcibly.",
backQuote(path),
e.nested() ? e.nested()->message() : e.message());
file.remove();
}
else
throw;
}
}
void DiskHDFS::removeMetaRecursive(const String & path, bool keep_in_remote_fs)
{
checkStackSize(); /// This is needed to prevent stack overflow in case of cyclic symlinks.
Poco::File file(metadata_path + path);
if (file.isFile())
{
removeMeta(path, keep_in_remote_fs);
}
else
{
for (auto it{iterateDirectory(path)}; it->isValid(); it->next())
removeMetaRecursive(it->path(), keep_in_remote_fs);
file.remove();
}
}
void registerDiskHDFS(DiskFactory & factory)
{
auto creator = [](const String & name,
@ -129,4 +231,5 @@ void registerDiskHDFS(DiskFactory & factory)
factory.registerDiskType("hdfs", creator);
}
}

16
src/Disks/HDFS/DiskHDFS.h
View File

@ -40,9 +40,23 @@ public:
std::unique_ptr<WriteBufferFromFileBase> writeFile(const String & path, size_t buf_size, WriteMode mode) override;
void removeFromRemoteFS(const Metadata & metadata) override;
void removeSharedFile(const String & path, bool keep_in_remote_fs) override;
void removeSharedRecursive(const String & path, bool) override;
void removeFileIfExists(const String & path) override;
void removeRecursive(const String & path) override;
void removeFile(const String & path) override { removeSharedFile(path, false); }
private:
void removeFromRemoteFS(const Metadata & metadata);
void removeMetaRecursive(const String & path, bool keep_in_remote_fs);
void removeMeta(const String & path, bool keep_in_remote_fs);
String getRandomName() { return toString(UUIDHelpers::generateV4()); }
const Poco::Util::AbstractConfiguration & config;

104
src/Disks/IDiskRemote.cpp
View File

@ -7,7 +7,6 @@
#include <IO/WriteBufferFromS3.h>
#include <IO/WriteHelpers.h>
#include <Poco/File.h>
#include <Common/checkStackSize.h>
#include <Common/createHardLink.h>
#include <Common/quoteString.h>
#include <common/logger_useful.h>
@ -19,10 +18,9 @@ namespace DB
namespace ErrorCodes
{
extern const int UNKNOWN_FORMAT;
extern const int INCORRECT_DISK_INDEX;
extern const int UNKNOWN_FORMAT;
extern const int FILE_ALREADY_EXISTS;
extern const int CANNOT_DELETE_DIRECTORY;
}
@ -151,69 +149,6 @@ IDiskRemote::Metadata IDiskRemote::createMeta(const String & path) const
}
void IDiskRemote::removeMeta(const String & path, bool keep_in_remote_fs)
{
Poco::File file(metadata_path + path);
if (!file.isFile())
throw Exception(ErrorCodes::CANNOT_DELETE_DIRECTORY, "Path '{}' is a directory", path);
try
{
auto metadata = readMeta(path);
/// If there is no references - delete content from S3.
if (metadata.ref_count == 0)
{
file.remove();
if (!keep_in_remote_fs)
removeFromRemoteFS(metadata);
}
else /// In other case decrement number of references, save metadata and delete file.
{
--metadata.ref_count;
metadata.save();
file.remove();
}
}
catch (const Exception & e)
{
/// If it's impossible to read meta - just remove it from FS.
if (e.code() == ErrorCodes::UNKNOWN_FORMAT)
{
LOG_WARNING(
log,
"Metadata file {} can't be read by reason: {}. Removing it forcibly.",
backQuote(path),
e.nested() ? e.nested()->message() : e.message());
file.remove();
}
else
throw;
}
}
void IDiskRemote::removeMetaRecursive(const String & path, bool keep_in_remote_fs)
{
checkStackSize(); /// This is needed to prevent stack overflow in case of cyclic symlinks.
Poco::File file(metadata_path + path);
if (file.isFile())
{
removeMeta(path, keep_in_remote_fs);
}
else
{
for (auto it{iterateDirectory(path)}; it->isValid(); it->next())
removeMetaRecursive(it->path(), keep_in_remote_fs);
file.remove();
}
}
DiskPtr DiskRemoteReservation::getDisk(size_t i) const
{
if (i != 0)
@ -323,43 +258,6 @@ void IDiskRemote::replaceFile(const String & from_path, const String & to_path)
}
void IDiskRemote::removeSharedFile(const String & path, bool keep_in_remote_fs)
{
removeMeta(path, keep_in_remote_fs);
}
void IDiskRemote::removeSharedRecursive(const String & path, bool keep_in_remote_fs)
{
removeMetaRecursive(path, keep_in_remote_fs);
}
void IDiskRemote::removeFileIfExists(const String & path)
{
if (Poco::File(metadata_path + path).exists())
removeMeta(path, /* keep_in_remote_fs */ false);
}
void IDiskRemote::removeRecursive(const String & path)
{
checkStackSize(); /// This is needed to prevent stack overflow in case of cyclic symlinks.
Poco::File file(metadata_path + path);
if (file.isFile())
{
removeFile(path);
}
else
{
for (auto it{iterateDirectory(path)}; it->isValid(); it->next())
removeRecursive(it->path());
file.remove();
}
}
void IDiskRemote::setReadOnly(const String & path)
{
/// We should store read only flag inside metadata file (instead of using FS flag),

16
src/Disks/IDiskRemote.h
View File

@ -48,16 +48,6 @@ public:
void createFile(const String & path) override;
void removeFile(const String & path) override { removeSharedFile(path, false); }
void removeSharedFile(const String & path, bool keep_in_remote_fs) override;
void removeSharedRecursive(const String & path, bool) override;
void removeFileIfExists(const String & path) override;
void removeRecursive(const String & path) override;
size_t getFileSize(const String & path) const override;
void moveFile(const String & from_path, const String & to_path) override;
@ -90,13 +80,7 @@ public:
ReservationPtr reserve(UInt64 bytes) override;
virtual void removeFromRemoteFS(const Metadata & /* metadata */) {}
protected:
void removeMetaRecursive(const String & path, bool keep_in_remote_fs);
void removeMeta(const String & path, bool keep_in_remote_fs);
const String disk_name;
const String remote_fs_root_path;
const String metadata_path;

1
src/Disks/S3/DiskS3.h
View File

@ -64,6 +64,7 @@ public:
size_t buf_size,
WriteMode mode) override;
void removeFile(const String & path) override { removeSharedFile(path, false); }
void removeFileIfExists(const String & path) override;
void removeRecursive(const String & path) override { removeSharedRecursive(path, false); }