: name(name_)

, copying_thread_pool(std::make_unique<ThreadPool>(

CurrentMetrics::IDiskCopierThreads,

CurrentMetrics::IDiskCopierThreadsActive,

CurrentMetrics::IDiskCopierThreadsScheduled,

config.getUInt(config_prefix + ".thread_pool_size", 16)))

{

: name(name_)

, copying_thread_pool(std::make_unique<ThreadPool>(

CurrentMetrics::IDiskCopierThreads, CurrentMetrics::IDiskCopierThreadsActive, CurrentMetrics::IDiskCopierThreadsScheduled, 16))

{

const String & from_file_path,

IDisk & to_disk,

const String & to_file_path,

const ReadSettings & read_settings,

const WriteSettings & write_settings,

const std::function<void()> & cancellation_hook

)

{

LOG_DEBUG(getLogger("IDisk"), "Copying from {} (path: {}) {} to {} (path: {}) {}.",

getName(), getPath(), from_file_path, to_disk.getName(), to_disk.getPath(), to_file_path);

auto in = readFile(from_file_path, read_settings);

auto out = to_disk.writeFile(to_file_path, DBMS_DEFAULT_BUFFER_SIZE, WriteMode::Rewrite, write_settings);

copyData(*in, *out, cancellation_hook);

out->finalize();

const String & path,

const ReadSettings & settings,

std::optional<size_t> read_hint) const

{

if (existsFile(path))

return readFile(path, settings, read_hint);

else

return {};

{

for (const auto & file : files)

{

bool keep_file = keep_all_batch_data || file_names_remove_metadata_only.contains(fs::path(file.path).filename());

if (file.if_exists)

removeSharedFileIfExists(file.path, keep_file);

else

removeSharedFile(file.path, keep_file);

}

IDisk & from_disk,

String from_path,

IDisk & to_disk,

String to_path,

ThreadPoolCallbackRunnerLocal<void> & runner,

const ReadSettings & read_settings,

const WriteSettings & write_settings,

const std::function<void()> & cancellation_hook)

{

if (from_disk.existsFile(from_path))

{

runner.enqueueAndKeepTrack(

[&from_disk, from_path, &to_disk, to_path, &read_settings, &write_settings, &cancellation_hook] {

from_disk.copyFile(

from_path, to_disk, to_path, read_settings, write_settings, cancellation_hook);

});

}

else /// Directory

{

fs::path dest(to_path);

to_disk.createDirectories(dest);

/// Calling asyncCopy recursively is fine here. Each call will capture by reference what were already references

/// dest is an exception, but it's passed as value, not reference

for (auto it = from_disk.iterateDirectory(from_path); it->isValid(); it->next())

asyncCopy(from_disk, it->path(), to_disk, dest / it->name(), runner, read_settings, write_settings, cancellation_hook);

}

const String & from_path,

const std::shared_ptr<IDisk> & to_disk,

const String & to_path,

const ReadSettings & read_settings,

WriteSettings write_settings,

const std::function<void()> & cancellation_hook)

{

ThreadPoolCallbackRunnerLocal<void> runner(*copying_thread_pool, ThreadName::ASYNC_COPY);

/// Disable parallel write. We already copy in parallel.

/// Avoid high memory usage. See test_s3_zero_copy_ttl/test.py::test_move_and_s3_memory_usage

write_settings.s3_allow_parallel_part_upload = false;

write_settings.azure_allow_parallel_part_upload = false;

/// This will capture by reference, which is fine since we got references ourselves and runner will be destroyed before returning

asyncCopy(*this, from_path, *to_disk, to_path, runner, read_settings, write_settings, cancellation_hook);

runner.waitForAllToFinishAndRethrowFirstError();

const String & from_dir,

const std::shared_ptr<IDisk> & to_disk,

const String & to_dir,

const ReadSettings & read_settings,

const WriteSettings & write_settings,

const std::function<void()> & cancellation_hook)

{

copyThroughBuffers(from_dir, to_disk, to_dir, read_settings, write_settings, cancellation_hook);

{

auto component_guard = Coordination::setCurrentComponent("IDisk::startup");

if (!skip_access_check)

{

if (isReadOnly())

{

LOG_DEBUG(getLogger("IDisk"),

"Skip access check for disk {} (read-only disk).",

getName());

}

else

checkAccess();

}

startupImpl();

{

DB::UUID server_uuid = DB::ServerUUID::get();

if (server_uuid == DB::UUIDHelpers::Nil)

throw Exception(ErrorCodes::LOGICAL_ERROR, "Server UUID is not initialized");

const String path = fmt::format("clickhouse_access_check_{}", toString(server_uuid));

checkAccessImpl(path);

try

{

const std::string_view payload("test", 4);

const auto read_settings = getReadSettings();

auto write_settings = getWriteSettings();

write_settings.is_initial_access_check = true;

/// write

{

auto file = writeFile(path, std::min<size_t>(DBMS_DEFAULT_BUFFER_SIZE, payload.size()), WriteMode::Rewrite, write_settings);

try

{

file->write(payload.data(), payload.size());

file->finalize();

}

catch (...)

{

/// Log current exception, because finalize() can throw a different exception.

tryLogCurrentException(__PRETTY_FUNCTION__);

throw;

}

}

/// read

{

auto file = readFile(path, read_settings);

String buf(payload.size(), '0');

file->readStrict(buf.data(), buf.size());

if (buf != payload)

{

throw Exception(ErrorCodes::CANNOT_READ_ALL_DATA,

"Content of {}::{} does not matches after read ({} vs {})", name, path, buf, payload);

}

}

/// read with offset

{

auto file = readFile(path, read_settings);

auto offset = 2;

String buf(payload.size() - offset, '0');

file->seek(offset, 0);

file->readStrict(buf.data(), buf.size());

if (buf != payload.substr(offset))

{

throw Exception(ErrorCodes::CANNOT_READ_ALL_DATA,

"Content of {}::{} does not matches after read with offset ({} vs {})", name, path, buf, payload.substr(offset));

}

}

/// check case sensitivity

{

std::unique_lock lock(case_sensitivity_check_mutex);

is_case_insensitive = existsFile(boost::to_upper_copy(path));

is_case_sensitivity_checked = true;

}

/// remove

removeFile(path);

}

catch (Exception & e)

{

e.addMessage(fmt::format("While checking access for disk {}", name));

throw;

{

/// For readonly disk we cannot perform the case sensitivity check.

if (isReadOnly())

throw Exception(ErrorCodes::NOT_IMPLEMENTED, "Cannot check if filesystem is case insensitive: disk {} is readonly", name);

if (is_case_sensitivity_checked)

return is_case_insensitive;

std::unique_lock lock(case_sensitivity_check_mutex);

const String path = fmt::format("clickhouse_case_sensitivity_check_{}", toString(DB::UUIDHelpers::generateV4()));

try

{

createFile(path);

is_case_insensitive = existsFile(boost::to_upper_copy(path));

is_case_sensitivity_checked = true;

removeFile(path);

}

catch (Exception & e)

{

e.addMessage(fmt::format("While checking case sensitivity for disk {}", name));

throw;

}

return is_case_insensitive;