thevar1able/ClickHouse
1
0
Fork 0
mirror of https://github.com/ClickHouse/ClickHouse.git synced 2024-09-19 16:20:50 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #64446 from divanik/divanik/fix_clickhouse_disks_2

Browse Source 
Interactive client for clickhouse-disks
This commit is contained in:
Daniil Ivanik 2024-07-08 09:00:15 +00:00 committed by GitHub
commit d00b12d0a9
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
36 changed files with 2070 additions and 732 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

61
docs/en/operations/utilities/clickhouse-disks.md
View File

@ -4,35 +4,56 @@ sidebar_position: 59
sidebar_label: clickhouse-disks
---
# clickhouse-disks
# Clickhouse-disks
A utility providing filesystem-like operations for ClickHouse disks.
A utility providing filesystem-like operations for ClickHouse disks. It can work in both interactive and not interactive modes.
Program-wide options:
## Program-wide options
* `--config-file, -C` -- path to ClickHouse config, defaults to `/etc/clickhouse-server/config.xml`.
* `--save-logs` -- Log progress of invoked commands to `/var/log/clickhouse-server/clickhouse-disks.log`.
* `--log-level` -- What [type](../server-configuration-parameters/settings#server_configuration_parameters-logger) of events to log, defaults to `none`.
* `--disk` -- what disk to use for `mkdir, move, read, write, remove` commands. Defaults to `default`.
* `--query, -q` -- single query that can be executed without launching interactive mode
* `--help, -h` -- print all the options and commands with description
## Default Disks
After the launch two disks are initialized. The first one is a disk `local` that is supposed to imitate local file system from which clickhouse-disks utility was launched. The second one is a disk `default` that is mounted to the local filesystem in the directory that can be found in config as a parameter `clickhouse/path` (default value is `/var/lib/clickhouse`).
## Clickhouse-disks state
For each disk that was added the utility stores current directory (as in a usual filesystem). User can change current directory and switch between disks.
State is reflected in a prompt "`disk_name`:`path_name`"
## Commands
* `copy [--disk-from d1] [--disk-to d2] <FROM_PATH> <TO_PATH>`.
Recursively copy data from `FROM_PATH` at disk `d1` (defaults to `disk` value if not provided)
to `TO_PATH` at disk `d2` (defaults to `disk` value if not provided).
* `move <FROM_PATH> <TO_PATH>`.
Move file or directory from `FROM_PATH` to `TO_PATH`.
* `remove <PATH>`.
Remove `PATH` recursively.
* `link <FROM_PATH> <TO_PATH>`.
Create a hardlink from `FROM_PATH` to `TO_PATH`.
* `list [--recursive] <PATH>...`
List files at `PATH`s. Non-recursive by default.
* `list-disks`.
In these documentation file all mandatory positional arguments are referred as `<parameter>`, named arguments are referred as `[--parameter value]`. All positional parameters could be mentioned as a named parameter with a corresponding name.
* `cd (change-dir, change_dir) [--disk disk] <path>`
Change directory to path `path` on disk `disk` (default value is a current disk). No disk switching happens.
* `copy (cp) [--disk-from disk_1] [--disk-to disk_2] <path-from> <path-to>`.
Recursively copy data from `path-from` at disk `disk_1` (default value is a current disk (parameter `disk` in a non-interactive mode))
to `path-to` at disk `disk_2` (default value is a current disk (parameter `disk` in a non-interactive mode)).
* `current_disk_with_path (current, current_disk, current_path)`
Print current state in format:
`Disk: "current_disk" Path: "current path on current disk"`
* `help [<command>]`
Print help message about command `command`. If `command` is not specified print information about all commands.
* `move (mv) <path-from> <path-to>`.
Move file or directory from `path-from` to `path-to` within current disk.
* `remove (rm, delete) <path>`.
Remove `path` recursively on a current disk.
* `link (ln) <path-from> <path-to>`.
Create a hardlink from `path-from` to `path-to` on a current disk.
* `list (ls) [--recursive] <path>`
List files at `path`s on a current disk. Non-recursive by default.
* `list-disks (list_disks, ls-disks, ls_disks)`.
List disks names.
* `mkdir [--recursive] <PATH>`.
* `mkdir [--recursive] <path>` on a current disk.
Create a directory. Non-recursive by default.
* `read: <FROM_PATH> [<TO_PATH>]`
Read a file from `FROM_PATH` to `TO_PATH` (`stdout` if not supplied).
* `write [FROM_PATH] <TO_PATH>`.
Write a file from `FROM_PATH` (`stdin` if not supplied) to `TO_PATH`.
* `read (r) <path-from> [--path-to path]`
Read a file from `path-from` to `path` (`stdout` if not supplied).
* `switch-disk [--path path] <disk>`
Switch to disk `disk` on path `path` (if `path` is not specified default value is a previous path on disk `disk`).
* `write (w) [--path-from path] <path-to>`.
Write a file from `path` (`stdin` if `path` is not supplied, input must finish by Ctrl+D) to `path-to`.

10
programs/disks/CMakeLists.txt
View File

@ -1,6 +1,8 @@
set (CLICKHOUSE_DISKS_SOURCES
DisksApp.cpp
DisksClient.cpp
ICommand.cpp
CommandChangeDirectory.cpp
CommandCopy.cpp
CommandLink.cpp
CommandList.cpp
@ -9,10 +11,14 @@ set (CLICKHOUSE_DISKS_SOURCES
CommandMove.cpp
CommandRead.cpp
CommandRemove.cpp
CommandWrite.cpp)
CommandSwitchDisk.cpp
CommandWrite.cpp
CommandHelp.cpp
CommandTouch.cpp
CommandGetCurrentDiskAndPath.cpp)
if (CLICKHOUSE_CLOUD)
set (CLICKHOUSE_DISKS_SOURCES ${CLICKHOUSE_DISKS_SOURCES} CommandPackedIO.cpp)
set (CLICKHOUSE_DISKS_SOURCES ${CLICKHOUSE_DISKS_SOURCES} CommandPackedIO.cpp)
endif ()
set (CLICKHOUSE_DISKS_LINK

35
programs/disks/CommandChangeDirectory.cpp Normal file
View File

@ -0,0 +1,35 @@
#include <Interpreters/Context.h>
#include <Common/TerminalSize.h>
#include "DisksApp.h"
#include "DisksClient.h"
#include "ICommand.h"
namespace DB
{
class CommandChangeDirectory final : public ICommand
{
public:
explicit CommandChangeDirectory() : ICommand()
{
command_name = "cd";
description = "Change directory (makes sense only in interactive mode)";
options_description.add_options()("path", po::value<String>(), "the path to which we want to change (mandatory, positional)")(
"disk", po::value<String>(), "A disk where the path is changed (without disk switching)");
positional_options_description.add("path", 1);
}
void executeImpl(const CommandLineOptions & options, DisksClient & client) override
{
DiskWithPath & disk = getDiskWithPath(client, options, "disk");
String path = getValueFromCommandLineOptionsThrow<String>(options, "path");
disk.setPath(path);
}
};
CommandPtr makeCommandChangeDirectory()
{
return std::make_shared<DB::CommandChangeDirectory>();
}
}

112
programs/disks/CommandCopy.cpp
View File

@ -1,6 +1,8 @@
#include "ICommand.h"
#include <Interpreters/Context.h>
#include "Common/Exception.h"
#include <Common/TerminalSize.h>
#include "DisksClient.h"
#include "ICommand.h"
namespace DB
{
@ -10,59 +12,89 @@ namespace ErrorCodes
extern const int BAD_ARGUMENTS;
}
class CommandCopy final : public ICommand
{
public:
CommandCopy()
explicit CommandCopy() : ICommand()
{
command_name = "copy";
command_option_description.emplace(createOptionsDescription("Allowed options", getTerminalWidth()));
description = "Recursively copy data from `FROM_PATH` to `TO_PATH`";
usage = "copy [OPTION]... <FROM_PATH> <TO_PATH>";
command_option_description->add_options()
("disk-from", po::value<String>(), "disk from which we copy")
("disk-to", po::value<String>(), "disk to which we copy");
description = "Recursively copy data from `path-from` to `path-to`";
options_description.add_options()(
"disk-from", po::value<String>(), "disk from which we copy is executed (default value is a current disk)")(
"disk-to", po::value<String>(), "disk to which copy is executed (default value is a current disk)")(
"path-from", po::value<String>(), "path from which copy is executed (mandatory, positional)")(
"path-to", po::value<String>(), "path to which copy is executed (mandatory, positional)")(
"recursive,r", "recursively copy the directory (required to remove a directory)");
positional_options_description.add("path-from", 1);
positional_options_description.add("path-to", 1);
}
void processOptions(
Poco::Util::LayeredConfiguration & config,
po::variables_map & options) const override
void executeImpl(const CommandLineOptions & options, DisksClient & client) override
{
if (options.count("disk-from"))
config.setString("disk-from", options["disk-from"].as<String>());
if (options.count("disk-to"))
config.setString("disk-to", options["disk-to"].as<String>());
}
auto disk_from = getDiskWithPath(client, options, "disk-from");
auto disk_to = getDiskWithPath(client, options, "disk-to");
String path_from = disk_from.getRelativeFromRoot(getValueFromCommandLineOptionsThrow<String>(options, "path-from"));
String path_to = disk_to.getRelativeFromRoot(getValueFromCommandLineOptionsThrow<String>(options, "path-to"));
bool recursive = options.count("recursive");
void execute(
const std::vector<String> & command_arguments,
std::shared_ptr<DiskSelector> & disk_selector,
Poco::Util::LayeredConfiguration & config) override
{
if (command_arguments.size() != 2)
if (!disk_from.getDisk()->exists(path_from))
{
printHelpMessage();
throw DB::Exception(DB::ErrorCodes::BAD_ARGUMENTS, "Bad Arguments");
throw Exception(
ErrorCodes::BAD_ARGUMENTS,
"cannot stat '{}' on disk '{}': No such file or directory",
path_from,
disk_from.getDisk()->getName());
}
else if (disk_from.getDisk()->isFile(path_from))
{
auto target_location = getTargetLocation(path_from, disk_to, path_to);
if (!disk_to.getDisk()->exists(target_location) || disk_to.getDisk()->isFile(target_location))
{
disk_from.getDisk()->copyFile(
path_from,
*disk_to.getDisk(),
target_location,
/* read_settings= */ {},
/* write_settings= */ {},
/* cancellation_hook= */ {});
}
else
{
throw Exception(
ErrorCodes::BAD_ARGUMENTS, "cannot overwrite directory {} with non-directory {}", target_location, path_from);
}
}
else if (disk_from.getDisk()->isDirectory(path_from))
{
if (!recursive)
{
throw Exception(ErrorCodes::BAD_ARGUMENTS, "--recursive not specified; omitting directory {}", path_from);
}
auto target_location = getTargetLocation(path_from, disk_to, path_to);
String disk_name_from = config.getString("disk-from", config.getString("disk", "default"));
String disk_name_to = config.getString("disk-to", config.getString("disk", "default"));
const String & path_from = command_arguments[0];
const String & path_to = command_arguments[1];
DiskPtr disk_from = disk_selector->get(disk_name_from);
DiskPtr disk_to = disk_selector->get(disk_name_to);
String relative_path_from = validatePathAndGetAsRelative(path_from);
String relative_path_to = validatePathAndGetAsRelative(path_to);
disk_from->copyDirectoryContent(relative_path_from, disk_to, relative_path_to, /* read_settings= */ {}, /* write_settings= */ {}, /* cancellation_hook= */ {});
if (disk_to.getDisk()->isFile(target_location))
{
throw Exception(ErrorCodes::BAD_ARGUMENTS, "cannot overwrite non-directory {} with directory {}", path_to, target_location);
}
else if (!disk_to.getDisk()->exists(target_location))
{
disk_to.getDisk()->createDirectory(target_location);
}
disk_from.getDisk()->copyDirectoryContent(
path_from,
disk_to.getDisk(),
target_location,
/* read_settings= */ {},
/* write_settings= */ {},
/* cancellation_hook= */ {});
}
}
};
CommandPtr makeCommandCopy()
{
return std::make_shared<DB::CommandCopy>();
}
std::unique_ptr <DB::ICommand> makeCommandCopy()
{
return std::make_unique<DB::CommandCopy>();
}

30
programs/disks/CommandGetCurrentDiskAndPath.cpp Normal file
View File

@ -0,0 +1,30 @@
#include <Interpreters/Context.h>
#include <Common/TerminalSize.h>
#include "DisksApp.h"
#include "DisksClient.h"
#include "ICommand.h"
namespace DB
{
class CommandGetCurrentDiskAndPath final : public ICommand
{
public:
explicit CommandGetCurrentDiskAndPath() : ICommand()
{
command_name = "current_disk_with_path";
description = "Prints current disk and path (which coincide with the prompt)";
}
void executeImpl(const CommandLineOptions &, DisksClient & client) override
{
auto disk = client.getCurrentDiskWithPath();
std::cout << "Disk: " << disk.getDisk()->getName() << "\nPath: " << disk.getCurrentPath() << std::endl;
}
};
CommandPtr makeCommandGetCurrentDiskAndPath()
{
return std::make_shared<DB::CommandGetCurrentDiskAndPath>();
}
}

43
programs/disks/CommandHelp.cpp Normal file
View File

@ -0,0 +1,43 @@
#include "DisksApp.h"
#include "ICommand.h"
#include <Interpreters/Context.h>
#include <Common/TerminalSize.h>
namespace DB
{
class CommandHelp final : public ICommand
{
public:
explicit CommandHelp(const DisksApp & disks_app_) : disks_app(disks_app_)
{
command_name = "help";
description = "Print help message about available commands";
options_description.add_options()(
"command", po::value<String>(), "A command to help with (optional, positional), if not specified, help lists all the commands");
positional_options_description.add("command", 1);
}
void executeImpl(const CommandLineOptions & options, DisksClient & /*client*/) override
{
std::optional<String> command = getValueFromCommandLineOptionsWithOptional<String>(options, "command");
if (command.has_value())
{
disks_app.printCommandHelpMessage(command.value());
}
else
{
disks_app.printAvailableCommandsHelpMessage();
}
}
const DisksApp & disks_app;
};
CommandPtr makeCommandHelp(const DisksApp & disks_app)
{
return std::make_shared<DB::CommandHelp>(disks_app);
}
}

50
programs/disks/CommandLink.cpp
View File

@ -1,14 +1,9 @@
#include "ICommand.h"
#include <Interpreters/Context.h>
#include "ICommand.h"
namespace DB
{
namespace ErrorCodes
{
extern const int BAD_ARGUMENTS;
}
class CommandLink final : public ICommand
{
public:
@ -16,42 +11,27 @@ public:
{
command_name = "link";
description = "Create hardlink from `from_path` to `to_path`";
usage = "link [OPTION]... <FROM_PATH> <TO_PATH>";
options_description.add_options()(
"path-from", po::value<String>(), "the path from which a hard link will be created (mandatory, positional)")(
"path-to", po::value<String>(), "the path where a hard link will be created (mandatory, positional)");
positional_options_description.add("path-from", 1);
positional_options_description.add("path-to", 1);
}
void processOptions(
Poco::Util::LayeredConfiguration &,
po::variables_map &) const override
void executeImpl(const CommandLineOptions & options, DisksClient & client) override
{
}
auto disk = client.getCurrentDiskWithPath();
void execute(
const std::vector<String> & command_arguments,
std::shared_ptr<DiskSelector> & disk_selector,
Poco::Util::LayeredConfiguration & config) override
{
if (command_arguments.size() != 2)
{
printHelpMessage();
throw DB::Exception(DB::ErrorCodes::BAD_ARGUMENTS, "Bad Arguments");
}
const String & path_from = disk.getRelativeFromRoot(getValueFromCommandLineOptionsThrow<String>(options, "path-from"));
const String & path_to = disk.getRelativeFromRoot(getValueFromCommandLineOptionsThrow<String>(options, "path-to"));
String disk_name = config.getString("disk", "default");
const String & path_from = command_arguments[0];
const String & path_to = command_arguments[1];
DiskPtr disk = disk_selector->get(disk_name);
String relative_path_from = validatePathAndGetAsRelative(path_from);
String relative_path_to = validatePathAndGetAsRelative(path_to);
disk->createHardLink(relative_path_from, relative_path_to);
disk.getDisk()->createHardLink(path_from, path_to);
}
};
CommandPtr makeCommandLink()
{
return std::make_shared<DB::CommandLink>();
}
std::unique_ptr <DB::ICommand> makeCommandLink()
{
return std::make_unique<DB::CommandLink>();
}

117
programs/disks/CommandList.cpp
View File

@ -1,98 +1,95 @@
#include "ICommand.h"
#include <Interpreters/Context.h>
#include <Common/TerminalSize.h>
#include "DisksApp.h"
#include "DisksClient.h"
#include "ICommand.h"
namespace DB
{
namespace ErrorCodes
{
extern const int BAD_ARGUMENTS;
}
class CommandList final : public ICommand
{
public:
CommandList()
explicit CommandList() : ICommand()
{
command_name = "list";
command_option_description.emplace(createOptionsDescription("Allowed options", getTerminalWidth()));
description = "List files at path[s]";
usage = "list [OPTION]... <PATH>...";
command_option_description->add_options()
("recursive", "recursively list all directories");
options_description.add_options()("recursive", "recursively list the directory")("all", "show hidden files")(
"path", po::value<String>(), "the path of listing (mandatory, positional)");
positional_options_description.add("path", 1);
}
void processOptions(
Poco::Util::LayeredConfiguration & config,
po::variables_map & options) const override
void executeImpl(const CommandLineOptions & options, DisksClient & client) override
{
if (options.count("recursive"))
config.setBool("recursive", true);
}
void execute(
const std::vector<String> & command_arguments,
std::shared_ptr<DiskSelector> & disk_selector,
Poco::Util::LayeredConfiguration & config) override
{
if (command_arguments.size() != 1)
{
printHelpMessage();
throw DB::Exception(DB::ErrorCodes::BAD_ARGUMENTS, "Bad Arguments");
}
String disk_name = config.getString("disk", "default");
const String & path = command_arguments[0];
DiskPtr disk = disk_selector->get(disk_name);
String relative_path = validatePathAndGetAsRelative(path);
bool recursive = config.getBool("recursive", false);
bool recursive = options.count("recursive");
bool show_hidden = options.count("all");
auto disk = client.getCurrentDiskWithPath();
String path = getValueFromCommandLineOptionsWithDefault<String>(options, "path", ".");
if (recursive)
listRecursive(disk, relative_path);
listRecursive(disk, path, show_hidden);
else
list(disk, relative_path);
list(disk, path, show_hidden);
}
private:
static void list(const DiskPtr & disk, const std::string & relative_path)
static void list(const DiskWithPath & disk, const std::string & path, bool show_hidden)
{
std::vector<String> file_names;
disk->listFiles(relative_path, file_names);
std::vector<String> file_names = disk.listAllFilesByPath(path);
std::vector<String> selected_and_sorted_file_names{};
for (const auto & file_name : file_names)
std::cout << file_name << '\n';
if (show_hidden || (!file_name.starts_with('.')))
selected_and_sorted_file_names.push_back(file_name);
std::sort(selected_and_sorted_file_names.begin(), selected_and_sorted_file_names.end());
for (const auto & file_name : selected_and_sorted_file_names)
{
std::cout << file_name << "\n";
}
}
static void listRecursive(const DiskPtr & disk, const std::string & relative_path)
static void listRecursive(const DiskWithPath & disk, const std::string & relative_path, bool show_hidden)
{
std::vector<String> file_names;
disk->listFiles(relative_path, file_names);
std::vector<String> file_names = disk.listAllFilesByPath(relative_path);
std::vector<String> selected_and_sorted_file_names{};
std::cout << relative_path << ":\n";
if (!file_names.empty())
{
for (const auto & file_name : file_names)
std::cout << file_name << '\n';
std::cout << "\n";
}
for (const auto & file_name : file_names)
if (show_hidden || (!file_name.starts_with('.')))
selected_and_sorted_file_names.push_back(file_name);
std::sort(selected_and_sorted_file_names.begin(), selected_and_sorted_file_names.end());
for (const auto & file_name : selected_and_sorted_file_names)
{
auto path = relative_path.empty() ? file_name : (relative_path + "/" + file_name);
if (disk->isDirectory(path))
listRecursive(disk, path);
std::cout << file_name << "\n";
}
std::cout << "\n";
for (const auto & file_name : selected_and_sorted_file_names)
{
auto path = [&]() -> String
{
if (relative_path.ends_with("/"))
{
return relative_path + file_name;
}
else
{
return relative_path + "/" + file_name;
}
}();
if (disk.isDirectory(path))
{
listRecursive(disk, path, show_hidden);
}
}
}
};
}
std::unique_ptr <DB::ICommand> makeCommandList()
CommandPtr makeCommandList()
{
return std::make_unique<DB::CommandList>();
return std::make_shared<DB::CommandList>();
}
}

66
programs/disks/CommandListDisks.cpp
View File

@ -1,68 +1,40 @@
#include "ICommand.h"
#include <algorithm>
#include <Interpreters/Context.h>
#include <Common/TerminalSize.h>
#include "DisksClient.h"
#include "ICommand.h"
namespace DB
{
namespace ErrorCodes
{
extern const int BAD_ARGUMENTS;
}
class CommandListDisks final : public ICommand
{
public:
CommandListDisks()
explicit CommandListDisks() : ICommand()
{
command_name = "list-disks";
description = "List disks names";
usage = "list-disks [OPTION]";
description = "Lists all available disks";
}
void processOptions(
Poco::Util::LayeredConfiguration &,
po::variables_map &) const override
{}
void execute(
const std::vector<String> & command_arguments,
std::shared_ptr<DiskSelector> &,
Poco::Util::LayeredConfiguration & config) override
void executeImpl(const CommandLineOptions &, DisksClient & client) override
{
if (!command_arguments.empty())
std::vector<String> sorted_and_selected{};
for (const auto & disk_name : client.getAllDiskNames())
{
printHelpMessage();
throw DB::Exception(DB::ErrorCodes::BAD_ARGUMENTS, "Bad Arguments");
sorted_and_selected.push_back(disk_name + ":" + client.getDiskWithPath(disk_name).getAbsolutePath(""));
}
constexpr auto config_prefix = "storage_configuration.disks";
constexpr auto default_disk_name = "default";
Poco::Util::AbstractConfiguration::Keys keys;
config.keys(config_prefix, keys);
bool has_default_disk = false;
/// For the output to be ordered
std::set<String> disks;
for (const auto & disk_name : keys)
std::sort(sorted_and_selected.begin(), sorted_and_selected.end());
for (const auto & disk_name : sorted_and_selected)
{
if (disk_name == default_disk_name)
has_default_disk = true;
disks.insert(disk_name);
std::cout << disk_name << "\n";
}
if (!has_default_disk)
disks.insert(default_disk_name);
for (const auto & disk : disks)
std::cout << disk << '\n';
}
};
}
std::unique_ptr <DB::ICommand> makeCommandListDisks()
private:
};
CommandPtr makeCommandListDisks()
{
return std::make_unique<DB::CommandListDisks>();
return std::make_shared<DB::CommandListDisks>();
}
}

54
programs/disks/CommandMkDir.cpp
View File

@ -6,61 +6,35 @@
namespace DB
{
namespace ErrorCodes
{
extern const int BAD_ARGUMENTS;
}
class CommandMkDir final : public ICommand
{
public:
CommandMkDir()
{
command_name = "mkdir";
command_option_description.emplace(createOptionsDescription("Allowed options", getTerminalWidth()));
description = "Create a directory";
usage = "mkdir [OPTION]... <PATH>";
command_option_description->add_options()
("recursive", "recursively create directories");
description = "Creates a directory";
options_description.add_options()("parents", "recursively create directories")(
"path", po::value<String>(), "the path on which directory should be created (mandatory, positional)");
positional_options_description.add("path", 1);
}
void processOptions(
Poco::Util::LayeredConfiguration & config,
po::variables_map & options) const override
void executeImpl(const CommandLineOptions & options, DisksClient & client) override
{
if (options.count("recursive"))
config.setBool("recursive", true);
}
bool recursive = options.count("parents");
auto disk = client.getCurrentDiskWithPath();
void execute(
const std::vector<String> & command_arguments,
std::shared_ptr<DiskSelector> & disk_selector,
Poco::Util::LayeredConfiguration & config) override
{
if (command_arguments.size() != 1)
{
printHelpMessage();
throw DB::Exception(DB::ErrorCodes::BAD_ARGUMENTS, "Bad Arguments");
}
String disk_name = config.getString("disk", "default");
const String & path = command_arguments[0];
DiskPtr disk = disk_selector->get(disk_name);
String relative_path = validatePathAndGetAsRelative(path);
bool recursive = config.getBool("recursive", false);
String path = disk.getRelativeFromRoot(getValueFromCommandLineOptionsThrow<String>(options, "path"));
if (recursive)
disk->createDirectories(relative_path);
disk.getDisk()->createDirectories(path);
else
disk->createDirectory(relative_path);
disk.getDisk()->createDirectory(path);
}
};
CommandPtr makeCommandMkDir()
{
return std::make_shared<DB::CommandMkDir>();
}
std::unique_ptr <DB::ICommand> makeCommandMkDir()
{
return std::make_unique<DB::CommandMkDir>();
}

83
programs/disks/CommandMove.cpp
View File

@ -1,5 +1,5 @@
#include "ICommand.h"
#include <Interpreters/Context.h>
#include "ICommand.h"
namespace DB
{
@ -9,6 +9,7 @@ namespace ErrorCodes
extern const int BAD_ARGUMENTS;
}
class CommandMove final : public ICommand
{
public:
@ -16,44 +17,62 @@ public:
{
command_name = "move";
description = "Move file or directory from `from_path` to `to_path`";
usage = "move [OPTION]... <FROM_PATH> <TO_PATH>";
options_description.add_options()("path-from", po::value<String>(), "path from which we copy (mandatory, positional)")(
"path-to", po::value<String>(), "path to which we copy (mandatory, positional)");
positional_options_description.add("path-from", 1);
positional_options_description.add("path-to", 1);
}
void processOptions(
Poco::Util::LayeredConfiguration &,
po::variables_map &) const override
{}
void execute(
const std::vector<String> & command_arguments,
std::shared_ptr<DiskSelector> & disk_selector,
Poco::Util::LayeredConfiguration & config) override
void executeImpl(const CommandLineOptions & options, DisksClient & client) override
{
if (command_arguments.size() != 2)
auto disk = client.getCurrentDiskWithPath();
String path_from = disk.getRelativeFromRoot(getValueFromCommandLineOptionsThrow<String>(options, "path-from"));
String path_to = disk.getRelativeFromRoot(getValueFromCommandLineOptionsThrow<String>(options, "path-to"));
if (disk.getDisk()->isFile(path_from))
{
printHelpMessage();
throw DB::Exception(DB::ErrorCodes::BAD_ARGUMENTS, "Bad Arguments");
disk.getDisk()->moveFile(path_from, path_to);
}
else if (disk.getDisk()->isDirectory(path_from))
{
auto target_location = getTargetLocation(path_from, disk, path_to);
if (!disk.getDisk()->exists(target_location))
{
disk.getDisk()->createDirectory(target_location);
disk.getDisk()->moveDirectory(path_from, target_location);
}
else
{
if (disk.getDisk()->isFile(target_location))
{
throw Exception(
ErrorCodes::BAD_ARGUMENTS, "cannot overwrite non-directory '{}' with directory '{}'", target_location, path_from);
}
if (!disk.getDisk()->isDirectoryEmpty(target_location))
{
throw Exception(ErrorCodes::BAD_ARGUMENTS, "cannot move '{}' to '{}': Directory not empty", path_from, target_location);
}
else
{
disk.getDisk()->moveDirectory(path_from, target_location);
}
}
}
else if (!disk.getDisk()->exists(path_from))
{
throw Exception(
ErrorCodes::BAD_ARGUMENTS,
"cannot stat '{}' on disk: '{}': No such file or directory",
path_from,
disk.getDisk()->getName());
}
String disk_name = config.getString("disk", "default");
const String & path_from = command_arguments[0];
const String & path_to = command_arguments[1];
DiskPtr disk = disk_selector->get(disk_name);
String relative_path_from = validatePathAndGetAsRelative(path_from);
String relative_path_to = validatePathAndGetAsRelative(path_to);
if (disk->isFile(relative_path_from))
disk->moveFile(relative_path_from, relative_path_to);
else
disk->moveDirectory(relative_path_from, relative_path_to);
}
};
CommandPtr makeCommandMove()
{
return std::make_shared<DB::CommandMove>();
}
std::unique_ptr <DB::ICommand> makeCommandMove()
{
return std::make_unique<DB::CommandMove>();
}

70
programs/disks/CommandRead.cpp
View File

@ -1,78 +1,52 @@
#include "ICommand.h"
#include <Interpreters/Context.h>
#include <IO/ReadBufferFromFile.h>
#include <IO/WriteBufferFromFile.h>
#include <IO/copyData.h>
#include <Interpreters/Context.h>
#include <Common/TerminalSize.h>
#include "ICommand.h"
namespace DB
{
namespace ErrorCodes
{
extern const int BAD_ARGUMENTS;
}
class CommandRead final : public ICommand
{
public:
CommandRead()
{
command_name = "read";
command_option_description.emplace(createOptionsDescription("Allowed options", getTerminalWidth()));
description = "Read a file from `FROM_PATH` to `TO_PATH`";
usage = "read [OPTION]... <FROM_PATH> [<TO_PATH>]";
command_option_description->add_options()
("output", po::value<String>(), "file to which we are reading, defaults to `stdout`");
description = "Read a file from `path-from` to `path-to`";
options_description.add_options()("path-from", po::value<String>(), "file from which we are reading (mandatory, positional)")(
"path-to", po::value<String>(), "file to which we are writing, defaults to `stdout`");
positional_options_description.add("path-from", 1);
}
void processOptions(
Poco::Util::LayeredConfiguration & config,
po::variables_map & options) const override
void executeImpl(const CommandLineOptions & options, DisksClient & client) override
{
if (options.count("output"))
config.setString("output", options["output"].as<String>());
}
auto disk = client.getCurrentDiskWithPath();
String path_from = disk.getRelativeFromRoot(getValueFromCommandLineOptionsThrow<String>(options, "path-from"));
std::optional<String> path_to = getValueFromCommandLineOptionsWithOptional<String>(options, "path-to");
void execute(
const std::vector<String> & command_arguments,
std::shared_ptr<DiskSelector> & disk_selector,
Poco::Util::LayeredConfiguration & config) override
{
if (command_arguments.size() != 1)
auto in = disk.getDisk()->readFile(path_from);
std::unique_ptr<WriteBufferFromFileBase> out = {};
if (path_to.has_value())
{
printHelpMessage();
throw DB::Exception(DB::ErrorCodes::BAD_ARGUMENTS, "Bad Arguments");
}
String disk_name = config.getString("disk", "default");
DiskPtr disk = disk_selector->get(disk_name);
String relative_path = validatePathAndGetAsRelative(command_arguments[0]);
String path_output = config.getString("output", "");
if (!path_output.empty())
{
String relative_path_output = validatePathAndGetAsRelative(path_output);
auto in = disk->readFile(relative_path);
auto out = disk->writeFile(relative_path_output);
String relative_path_to = disk.getRelativeFromRoot(path_to.value());
out = disk.getDisk()->writeFile(relative_path_to);
copyData(*in, *out);
out->finalize();
}
else
{
auto in = disk->readFile(relative_path);
std::unique_ptr<WriteBufferFromFileBase> out = std::make_unique<WriteBufferFromFileDescriptor>(STDOUT_FILENO);
out = std::make_unique<WriteBufferFromFileDescriptor>(STDOUT_FILENO);
copyData(*in, *out);
out->write('\n');
}
out->finalize();
}
};
CommandPtr makeCommandRead()
{
return std::make_shared<DB::CommandRead>();
}
std::unique_ptr <DB::ICommand> makeCommandRead()
{
return std::make_unique<DB::CommandRead>();
}

60
programs/disks/CommandRemove.cpp
View File

@ -1,5 +1,6 @@
#include "ICommand.h"
#include <Interpreters/Context.h>
#include "Common/Exception.h"
#include "ICommand.h"
namespace DB
{
@ -9,46 +10,49 @@ namespace ErrorCodes
extern const int BAD_ARGUMENTS;
}
class CommandRemove final : public ICommand
{
public:
CommandRemove()
{
command_name = "remove";
description = "Remove file or directory with all children. Throws exception if file doesn't exists.\nPath should be in format './' or './path' or 'path'";
usage = "remove [OPTION]... <PATH>";
description = "Remove file or directory. Throws exception if file doesn't exists";
options_description.add_options()("path", po::value<String>(), "path that is going to be deleted (mandatory, positional)")(
"recursive,r", "recursively removes the directory (required to remove a directory)");
positional_options_description.add("path", 1);
}
void processOptions(
Poco::Util::LayeredConfiguration &,
po::variables_map &) const override
{}
void execute(
const std::vector<String> & command_arguments,
std::shared_ptr<DiskSelector> & disk_selector,
Poco::Util::LayeredConfiguration & config) override
void executeImpl(const CommandLineOptions & options, DisksClient & client) override
{
if (command_arguments.size() != 1)
auto disk = client.getCurrentDiskWithPath();
const String & path = disk.getRelativeFromRoot(getValueFromCommandLineOptionsThrow<String>(options, "path"));
bool recursive = options.count("recursive");
if (!disk.getDisk()->exists(path))
{
printHelpMessage();
throw DB::Exception(DB::ErrorCodes::BAD_ARGUMENTS, "Bad Arguments");
throw Exception(ErrorCodes::BAD_ARGUMENTS, "Path {} on disk {} doesn't exist", path, disk.getDisk()->getName());
}
else if (disk.getDisk()->isDirectory(path))
{
if (!recursive)
{
throw Exception(ErrorCodes::BAD_ARGUMENTS, "cannot remove '{}': Is a directory", path);
}
else
{
disk.getDisk()->removeRecursive(path);
}
}
else
{
disk.getDisk()->removeFileIfExists(path);
}
String disk_name = config.getString("disk", "default");
const String & path = command_arguments[0];
DiskPtr disk = disk_selector->get(disk_name);
String relative_path = validatePathAndGetAsRelative(path);
disk->removeRecursive(relative_path);
}
};
CommandPtr makeCommandRemove()
{
return std::make_shared<DB::CommandRemove>();
}
std::unique_ptr <DB::ICommand> makeCommandRemove()
{
return std::make_unique<DB::CommandRemove>();
}

35
programs/disks/CommandSwitchDisk.cpp Normal file
View File

@ -0,0 +1,35 @@
#include <optional>
#include <Interpreters/Context.h>
#include <Common/TerminalSize.h>
#include "DisksApp.h"
#include "ICommand.h"
namespace DB
{
class CommandSwitchDisk final : public ICommand
{
public:
explicit CommandSwitchDisk() : ICommand()
{
command_name = "switch-disk";
description = "Switch disk (makes sense only in interactive mode)";
options_description.add_options()("disk", po::value<String>(), "the disk to switch to (mandatory, positional)")(
"path", po::value<String>(), "the path to switch on the disk");
positional_options_description.add("disk", 1);
}
void executeImpl(const CommandLineOptions & options, DisksClient & client) override
{
String disk = getValueFromCommandLineOptions<String>(options, "disk");
std::optional<String> path = getValueFromCommandLineOptionsWithOptional<String>(options, "path");
client.switchToDisk(disk, path);
}
};
CommandPtr makeCommandSwitchDisk()
{
return std::make_shared<DB::CommandSwitchDisk>();
}
}

34
programs/disks/CommandTouch.cpp Normal file
View File

@ -0,0 +1,34 @@
#include <Interpreters/Context.h>
#include <Common/TerminalSize.h>
#include "DisksApp.h"
#include "DisksClient.h"
#include "ICommand.h"
namespace DB
{
class CommandTouch final : public ICommand
{
public:
explicit CommandTouch() : ICommand()
{
command_name = "touch";
description = "Create a file by path";
options_description.add_options()("path", po::value<String>(), "the path of listing (mandatory, positional)");
positional_options_description.add("path", 1);
}
void executeImpl(const CommandLineOptions & options, DisksClient & client) override
{
auto disk = client.getCurrentDiskWithPath();
String path = getValueFromCommandLineOptionsThrow<String>(options, "path");
disk.getDisk()->createFile(disk.getRelativeFromRoot(path));
}
};
CommandPtr makeCommandTouch()
{
return std::make_shared<DB::CommandTouch>();
}
}

80
programs/disks/CommandWrite.cpp
View File

@ -1,79 +1,57 @@
#include "ICommand.h"
#include <Interpreters/Context.h>
#include "ICommand.h"
#include <Common/TerminalSize.h>
#include <IO/ReadBufferFromFile.h>
#include <IO/WriteBufferFromFile.h>
#include <IO/copyData.h>
#include <Common/TerminalSize.h>
namespace DB
{
namespace ErrorCodes
{
extern const int BAD_ARGUMENTS;
}
class CommandWrite final : public ICommand
{
public:
CommandWrite()
{
command_name = "write";
command_option_description.emplace(createOptionsDescription("Allowed options", getTerminalWidth()));
description = "Write a file from `FROM_PATH` to `TO_PATH`";
usage = "write [OPTION]... [<FROM_PATH>] <TO_PATH>";
command_option_description->add_options()
("input", po::value<String>(), "file from which we are reading, defaults to `stdin`");
description = "Write a file from `path-from` to `path-to`";
options_description.add_options()("path-from", po::value<String>(), "file from which we are reading, defaults to `stdin` (input from `stdin` is finished by Ctrl+D)")(
"path-to", po::value<String>(), "file to which we are writing (mandatory, positional)");
positional_options_description.add("path-to", 1);
}
void processOptions(
Poco::Util::LayeredConfiguration & config,
po::variables_map & options) const override
void executeImpl(const CommandLineOptions & options, DisksClient & client) override
{
if (options.count("input"))
config.setString("input", options["input"].as<String>());
}
auto disk = client.getCurrentDiskWithPath();
void execute(
const std::vector<String> & command_arguments,
std::shared_ptr<DiskSelector> & disk_selector,
Poco::Util::LayeredConfiguration & config) override
{
if (command_arguments.size() != 1)
std::optional<String> path_from = getValueFromCommandLineOptionsWithOptional<String>(options, "path-from");
String path_to = disk.getRelativeFromRoot(getValueFromCommandLineOptionsThrow<String>(options, "path-to"));
auto in = [&]() -> std::unique_ptr<ReadBufferFromFileBase>
{
printHelpMessage();
throw DB::Exception(DB::ErrorCodes::BAD_ARGUMENTS, "Bad Arguments");
}
if (!path_from.has_value())
{
return std::make_unique<ReadBufferFromFileDescriptor>(STDIN_FILENO);
}
else
{
String relative_path_from = disk.getRelativeFromRoot(path_from.value());
return disk.getDisk()->readFile(relative_path_from);
}
}();
String disk_name = config.getString("disk", "default");
const String & path = command_arguments[0];
DiskPtr disk = disk_selector->get(disk_name);
String relative_path = validatePathAndGetAsRelative(path);
String path_input = config.getString("input", "");
std::unique_ptr<ReadBufferFromFileBase> in;
if (path_input.empty())
{
in = std::make_unique<ReadBufferFromFileDescriptor>(STDIN_FILENO);
}
else
{
String relative_path_input = validatePathAndGetAsRelative(path_input);
in = disk->readFile(relative_path_input);
}
auto out = disk->writeFile(relative_path);
auto out = disk.getDisk()->writeFile(path_to);
copyData(*in, *out);
out->finalize();
}
};
CommandPtr makeCommandWrite()
{
return std::make_shared<DB::CommandWrite>();
}
std::unique_ptr <DB::ICommand> makeCommandWrite()
{
return std::make_unique<DB::CommandWrite>();
}

547
programs/disks/DisksApp.cpp
View File

@ -1,11 +1,22 @@
#include "DisksApp.h"
#include <Client/ClientBase.h>
#include <Client/ReplxxLineReader.h>
#include "Common/Exception.h"
#include "Common/filesystemHelpers.h"
#include <Common/Config/ConfigProcessor.h>
#include "DisksClient.h"
#include "ICommand.h"
#include "ICommand_fwd.h"
#include <cstring>
#include <filesystem>
#include <memory>
#include <optional>
#include <Disks/registerDisks.h>
#include <Common/TerminalSize.h>
#include <Formats/registerFormats.h>
#include <Common/TerminalSize.h>
namespace DB
{
@ -13,74 +24,289 @@ namespace DB
namespace ErrorCodes
{
extern const int BAD_ARGUMENTS;
extern const int LOGICAL_ERROR;
};
LineReader::Patterns DisksApp::query_extenders = {"\\"};
LineReader::Patterns DisksApp::query_delimiters = {""};
String DisksApp::word_break_characters = " \t\v\f\a\b\r\n";
CommandPtr DisksApp::getCommandByName(const String & command) const
{
try
{
if (auto it = aliases.find(command); it != aliases.end())
return command_descriptions.at(it->second);
return command_descriptions.at(command);
}
catch (std::out_of_range &)
{
throw Exception(ErrorCodes::BAD_ARGUMENTS, "The command `{}` is unknown", command);
}
}
size_t DisksApp::findCommandPos(std::vector<String> & common_arguments)
std::vector<String> DisksApp::getEmptyCompletion(String command_name) const
{
for (size_t i = 0; i < common_arguments.size(); i++)
if (supported_commands.contains(common_arguments[i]))
return i + 1;
return common_arguments.size();
auto command_ptr = command_descriptions.at(command_name);
std::vector<String> answer{};
if (multidisk_commands.contains(command_ptr->command_name))
{
answer = client->getAllFilesByPatternFromAllDisks("");
}
else
{
answer = client->getCurrentDiskWithPath().getAllFilesByPattern("");
}
for (const auto & disk_name : client->getAllDiskNames())
{
answer.push_back(disk_name);
}
for (const auto & option : command_ptr->options_description.options())
{
answer.push_back("--" + option->long_name());
}
if (command_name == "help")
{
for (const auto & [current_command_name, description] : command_descriptions)
{
answer.push_back(current_command_name);
}
}
std::sort(answer.begin(), answer.end());
return answer;
}
void DisksApp::printHelpMessage(ProgramOptionsDescription & command_option_description)
std::vector<String> DisksApp::getCommandsToComplete(const String & command_prefix) const
{
std::optional<ProgramOptionsDescription> help_description =
createOptionsDescription("Help Message for clickhouse-disks", getTerminalWidth());
help_description->add(command_option_description);
std::cout << "ClickHouse disk management tool\n";
std::cout << "Usage: ./clickhouse-disks [OPTION]\n";
std::cout << "clickhouse-disks\n\n";
for (const auto & current_command : supported_commands)
std::cout << command_descriptions[current_command]->command_name
<< "\t"
<< command_descriptions[current_command]->description
<< "\n\n";
std::cout << command_option_description << '\n';
std::vector<String> answer{};
for (const auto & [word, _] : command_descriptions)
{
if (word.starts_with(command_prefix))
{
answer.push_back(word);
}
}
if (!answer.empty())
{
std::sort(answer.begin(), answer.end());
return answer;
}
for (const auto & [word, _] : aliases)
{
if (word.starts_with(command_prefix))
{
answer.push_back(word);
}
}
if (!answer.empty())
{
std::sort(answer.begin(), answer.end());
return answer;
}
return {command_prefix};
}
String DisksApp::getDefaultConfigFileName()
std::vector<String> DisksApp::getCompletions(const String & prefix) const
{
return "/etc/clickhouse-server/config.xml";
auto arguments = po::split_unix(prefix, word_break_characters);
if (arguments.empty())
{
return {};
}
if (word_break_characters.contains(prefix.back()))
{
CommandPtr command;
try
{
command = getCommandByName(arguments[0]);
}
catch (...)
{
return {arguments.back()};
}
return getEmptyCompletion(command->command_name);
}
else if (arguments.size() == 1)
{
String command_prefix = arguments[0];
return getCommandsToComplete(command_prefix);
}
else
{
String last_token = arguments.back();
CommandPtr command;
try
{
command = getCommandByName(arguments[0]);
}
catch (...)
{
return {last_token};
}
std::vector<String> answer = {};
if (command->command_name == "help")
{
return getCommandsToComplete(last_token);
}
else
{
answer = [&]() -> std::vector<String>
{
if (multidisk_commands.contains(command->command_name))
{
return client->getAllFilesByPatternFromAllDisks(last_token);
}
else
{
return client->getCurrentDiskWithPath().getAllFilesByPattern(last_token);
}
}();
for (const auto & disk_name : client->getAllDiskNames())
{
if (disk_name.starts_with(last_token))
{
answer.push_back(disk_name);
}
}
for (const auto & option : command->options_description.options())
{
String option_sign = "--" + option->long_name();
if (option_sign.starts_with(last_token))
{
answer.push_back(option_sign);
}
}
}
if (!answer.empty())
{
std::sort(answer.begin(), answer.end());
return answer;
}
else
{
return {last_token};
}
}
}
void DisksApp::addOptions(
ProgramOptionsDescription & options_description_,
boost::program_options::positional_options_description & positional_options_description
)
bool DisksApp::processQueryText(const String & text)
{
options_description_.add_options()
("help,h", "Print common help message")
("config-file,C", po::value<String>(), "Set config file")
("disk", po::value<String>(), "Set disk name")
("command_name", po::value<String>(), "Name for command to do")
("save-logs", "Save logs to a file")
("log-level", po::value<String>(), "Logging level")
;
if (text.find_first_not_of(word_break_characters) == std::string::npos)
{
return true;
}
if (exit_strings.find(text) != exit_strings.end())
return false;
CommandPtr command;
try
{
auto arguments = po::split_unix(text, word_break_characters);
command = getCommandByName(arguments[0]);
arguments.erase(arguments.begin());
command->execute(arguments, *client);
}
catch (DB::Exception & err)
{
int code = getCurrentExceptionCode();
if (code == ErrorCodes::LOGICAL_ERROR)
{
throw std::move(err);
}
else if (code == ErrorCodes::BAD_ARGUMENTS)
{
std::cerr << err.message() << "\n"
<< "\n";
if (command.get())
{
std::cerr << "COMMAND: " << command->command_name << "\n";
std::cerr << command->options_description << "\n";
}
else
{
printAvailableCommandsHelpMessage();
}
}
else
{
std::cerr << err.message() << "\n";
}
}
catch (std::exception & err)
{
std::cerr << err.what() << "\n";
}
positional_options_description.add("command_name", 1);
return true;
}
supported_commands = {"list-disks", "list", "move", "remove", "link", "copy", "write", "read", "mkdir"};
#ifdef CLICKHOUSE_CLOUD
supported_commands.insert("packed-io");
#endif
void DisksApp::runInteractiveReplxx()
{
ReplxxLineReader lr(
suggest,
history_file,
/* multiline= */ false,
query_extenders,
query_delimiters,
word_break_characters.c_str(),
/* highlighter_= */ {});
lr.enableBracketedPaste();
while (true)
{
DiskWithPath disk_with_path = client->getCurrentDiskWithPath();
String prompt = "\x1b[1;34m" + disk_with_path.getDisk()->getName() + "\x1b[0m:" + "\x1b[1;31m" + disk_with_path.getCurrentPath()
+ "\x1b[0m$ ";
auto input = lr.readLine(prompt, "\x1b[1;31m:-] \x1b[0m");
if (input.empty())
break;
if (!processQueryText(input))
break;
}
}
void DisksApp::parseAndCheckOptions(
const std::vector<String> & arguments, const ProgramOptionsDescription & options_description, CommandLineOptions & options)
{
auto parser = po::command_line_parser(arguments).options(options_description).allow_unregistered();
po::parsed_options parsed = parser.run();
po::store(parsed, options);
}
void DisksApp::addOptions()
{
options_description.add_options()("help,h", "Print common help message")("config-file,C", po::value<String>(), "Set config file")(
"disk", po::value<String>(), "Set disk name")("save-logs", "Save logs to a file")(
"log-level", po::value<String>(), "Logging level")("query,q", po::value<String>(), "Query for a non-interactive mode")(
"test-mode", "Interactive interface in test regyme");
command_descriptions.emplace("list-disks", makeCommandListDisks());
command_descriptions.emplace("copy", makeCommandCopy());
command_descriptions.emplace("list", makeCommandList());
command_descriptions.emplace("cd", makeCommandChangeDirectory());
command_descriptions.emplace("move", makeCommandMove());
command_descriptions.emplace("remove", makeCommandRemove());
command_descriptions.emplace("link", makeCommandLink());
command_descriptions.emplace("copy", makeCommandCopy());
command_descriptions.emplace("write", makeCommandWrite());
command_descriptions.emplace("read", makeCommandRead());
command_descriptions.emplace("mkdir", makeCommandMkDir());
command_descriptions.emplace("switch-disk", makeCommandSwitchDisk());
command_descriptions.emplace("current_disk_with_path", makeCommandGetCurrentDiskAndPath());
command_descriptions.emplace("touch", makeCommandTouch());
command_descriptions.emplace("help", makeCommandHelp(*this));
#ifdef CLICKHOUSE_CLOUD
command_descriptions.emplace("packed-io", makeCommandPackedIO());
#endif
for (const auto & [command_name, command_ptr] : command_descriptions)
{
if (command_name != command_ptr->command_name)
{
throw Exception(ErrorCodes::LOGICAL_ERROR, "Command name inside map doesn't coincide with actual command name");
}
}
}
void DisksApp::processOptions()
@ -93,76 +319,122 @@ void DisksApp::processOptions()
config().setBool("save-logs", true);
if (options.count("log-level"))
config().setString("log-level", options["log-level"].as<String>());
if (options.count("test-mode"))
config().setBool("test-mode", true);
if (options.count("query"))
query = std::optional{options["query"].as<String>()};
}
DisksApp::~DisksApp()
void DisksApp::printEntryHelpMessage() const
{
if (global_context)
global_context->shutdown();
std::cout << "\x1b[1;33m ClickHouse disk management tool \x1b[0m \n";
std::cout << options_description << '\n';
}
void DisksApp::init(std::vector<String> & common_arguments)
void DisksApp::printAvailableCommandsHelpMessage() const
{
stopOptionsProcessing();
std::cout << "\x1b[1;32mAvailable commands:\x1b[0m\n";
std::vector<std::pair<String, CommandPtr>> commands_with_aliases_and_descrtiptions{};
size_t maximal_command_length = 0;
for (const auto & [command_name, command_ptr] : command_descriptions)
{
std::string command_string = getCommandLineWithAliases(command_ptr);
maximal_command_length = std::max(maximal_command_length, command_string.size());
commands_with_aliases_and_descrtiptions.push_back({std::move(command_string), command_descriptions.at(command_name)});
}
for (const auto & [command_with_aliases, command_ptr] : commands_with_aliases_and_descrtiptions)
{
std::cout << "\x1b[1;33m" << command_with_aliases << "\x1b[0m" << std::string(5, ' ') << "\x1b[1;33m" << command_ptr->description
<< "\x1b[0m \n";
std::cout << command_ptr->options_description;
std::cout << std::endl;
}
}
ProgramOptionsDescription options_description{createOptionsDescription("clickhouse-disks", getTerminalWidth())};
void DisksApp::printCommandHelpMessage(CommandPtr command) const
{
String command_name_with_aliases = getCommandLineWithAliases(command);
std::cout << "\x1b[1;32m" << command_name_with_aliases << "\x1b[0m" << std::string(2, ' ') << command->description << "\n";
std::cout << command->options_description;
}
po::positional_options_description positional_options_description;
void DisksApp::printCommandHelpMessage(String command_name) const
{
printCommandHelpMessage(getCommandByName(command_name));
}
addOptions(options_description, positional_options_description);
String DisksApp::getCommandLineWithAliases(CommandPtr command) const
{
String command_string = command->command_name;
bool need_comma = false;
for (const auto & [alias_name, alias_command_name] : aliases)
{
if (alias_command_name == command->command_name)
{
if (std::exchange(need_comma, true))
command_string += ",";
else
command_string += "(";
command_string += alias_name;
}
}
command_string += (need_comma ? ")" : "");
return command_string;
}
size_t command_pos = findCommandPos(common_arguments);
std::vector<String> global_flags(command_pos);
command_arguments.resize(common_arguments.size() - command_pos);
copy(common_arguments.begin(), common_arguments.begin() + command_pos, global_flags.begin());
copy(common_arguments.begin() + command_pos, common_arguments.end(), command_arguments.begin());
void DisksApp::initializeHistoryFile()
{
String home_path;
const char * home_path_cstr = getenv("HOME"); // NOLINT(concurrency-mt-unsafe)
if (home_path_cstr)
home_path = home_path_cstr;
if (config().has("history-file"))
history_file = config().getString("history-file");
else
history_file = home_path + "/.disks-file-history";
parseAndCheckOptions(options_description, positional_options_description, global_flags);
if (!history_file.empty() && !fs::exists(history_file))
{
try
{
FS::createFile(history_file);
}
catch (const ErrnoException & e)
{
if (e.getErrno() != EEXIST)
throw;
}
}
}
void DisksApp::init(const std::vector<String> & common_arguments)
{
addOptions();
parseAndCheckOptions(common_arguments, options_description, options);
po::notify(options);
if (options.count("help"))
{
printHelpMessage(options_description);
printEntryHelpMessage();
printAvailableCommandsHelpMessage();
exit(0); // NOLINT(concurrency-mt-unsafe)
}
if (!supported_commands.contains(command_name))
{
std::cerr << "Unknown command name: " << command_name << "\n";
printHelpMessage(options_description);
throw DB::Exception(DB::ErrorCodes::BAD_ARGUMENTS, "Bad Arguments");
}
processOptions();
}
void DisksApp::parseAndCheckOptions(
ProgramOptionsDescription & options_description_,
boost::program_options::positional_options_description & positional_options_description,
std::vector<String> & arguments)
String DisksApp::getDefaultConfigFileName()
{
auto parser = po::command_line_parser(arguments)
.options(options_description_)
.positional(positional_options_description)
.allow_unregistered();
po::parsed_options parsed = parser.run();
po::store(parsed, options);
auto positional_arguments = po::collect_unrecognized(parsed.options, po::collect_unrecognized_mode::include_positional);
for (const auto & arg : positional_arguments)
{
if (command_descriptions.contains(arg))
{
command_name = arg;
break;
}
}
return "/etc/clickhouse-server/config.xml";
}
int DisksApp::main(const std::vector<String> & /*args*/)
{
std::vector<std::string> keys;
config().keys(keys);
if (config().has("config-file") || fs::exists(getDefaultConfigFileName()))
{
String config_path = config().getString("config-file", getDefaultConfigFileName());
@ -173,9 +445,13 @@ int DisksApp::main(const std::vector<String> & /*args*/)
}
else
{
printEntryHelpMessage();
throw Exception(ErrorCodes::BAD_ARGUMENTS, "No config-file specified");
}
config().keys(keys);
initializeHistoryFile();
if (config().has("save-logs"))
{
auto log_level = config().getString("log-level", "trace");
@ -200,61 +476,68 @@ int DisksApp::main(const std::vector<String> & /*args*/)
global_context->setApplicationType(Context::ApplicationType::DISKS);
String path = config().getString("path", DBMS_DEFAULT_PATH);
global_context->setPath(path);
auto & command = command_descriptions[command_name];
String main_disk = config().getString("disk", "default");
auto command_options = command->getCommandOptions();
std::vector<String> args;
if (command_options)
auto validator = [](const Poco::Util::AbstractConfiguration &, const std::string &, const std::string &) { return true; };
constexpr auto config_prefix = "storage_configuration.disks";
auto disk_selector = std::make_shared<DiskSelector>(std::unordered_set<String>{"cache", "encrypted"});
disk_selector->initialize(config(), config_prefix, global_context, validator);
std::vector<std::pair<DiskPtr, std::optional<String>>> disks_with_path;
for (const auto & [_, disk_ptr] : disk_selector->getDisksMap())
{
auto parser = po::command_line_parser(command_arguments).options(*command_options).allow_unregistered();
po::parsed_options parsed = parser.run();
po::store(parsed, options);
po::notify(options);
disks_with_path.emplace_back(
disk_ptr, (disk_ptr->getName() == "local") ? std::optional{fs::current_path().string()} : std::nullopt);
}
args = po::collect_unrecognized(parsed.options, po::collect_unrecognized_mode::include_positional);
command->processOptions(config(), options);
client = std::make_unique<DisksClient>(std::move(disks_with_path), main_disk);
suggest.setCompletionsCallback([&](const String & prefix, size_t /* prefix_length */) { return getCompletions(prefix); });
if (!query.has_value())
{
runInteractive();
}
else
{
auto parser = po::command_line_parser(command_arguments).options({}).allow_unregistered();
po::parsed_options parsed = parser.run();
args = po::collect_unrecognized(parsed.options, po::collect_unrecognized_mode::include_positional);
processQueryText(query.value());
}
std::unordered_set<std::string> disks
{
config().getString("disk", "default"),
config().getString("disk-from", config().getString("disk", "default")),
config().getString("disk-to", config().getString("disk", "default")),
};
auto validator = [&disks](
const Poco::Util::AbstractConfiguration & config,
const std::string & disk_config_prefix,
const std::string & disk_name)
{
if (!disks.contains(disk_name))
return false;
const auto disk_type = config.getString(disk_config_prefix + ".type", "local");
if (disk_type == "cache")
throw Exception(ErrorCodes::BAD_ARGUMENTS, "Disk type 'cache' of disk {} is not supported by clickhouse-disks", disk_name);
return true;
};
constexpr auto config_prefix = "storage_configuration.disks";
auto disk_selector = std::make_shared<DiskSelector>();
disk_selector->initialize(config(), config_prefix, global_context, validator);
command->execute(args, disk_selector, config());
return Application::EXIT_OK;
}
DisksApp::~DisksApp()
{
client.reset(nullptr);
if (global_context)
global_context->shutdown();
}
void DisksApp::runInteractiveTestMode()
{
for (String input; std::getline(std::cin, input);)
{
if (!processQueryText(input))
break;
std::cout << "\a\a\a\a" << std::endl;
std::cerr << std::flush;
}
}
void DisksApp::runInteractive()
{
if (config().hasOption("test-mode"))
runInteractiveTestMode();
else
runInteractiveReplxx();
}
}
int mainEntryClickHouseDisks(int argc, char ** argv)
@ -269,16 +552,16 @@ int mainEntryClickHouseDisks(int argc, char ** argv)
catch (const DB::Exception & e)
{
std::cerr << DB::getExceptionMessage(e, false) << std::endl;
return 1;
return 0;
}
catch (const boost::program_options::error & e)
{
std::cerr << "Bad arguments: " << e.what() << std::endl;
return DB::ErrorCodes::BAD_ARGUMENTS;
return 0;
}
catch (...)
{
std::cerr << DB::getCurrentExceptionMessage(true) << std::endl;
return 1;
return 0;
}
}

114
programs/disks/DisksApp.h
View File

@ -1,61 +1,107 @@
#pragma once
#include <unordered_map>
#include <vector>
#include <Client/ReplxxLineReader.h>
#include <Loggers/Loggers.h>
#include "DisksClient.h"
#include "ICommand_fwd.h"
#include <Interpreters/Context.h>
#include <boost/program_options/options_description.hpp>
#include <boost/program_options/variables_map.hpp>
#include <Poco/Util/Application.h>
#include <boost/program_options.hpp>
namespace DB
{
class ICommand;
using CommandPtr = std::unique_ptr<ICommand>;
namespace po = boost::program_options;
using ProgramOptionsDescription = boost::program_options::options_description;
using CommandLineOptions = boost::program_options::variables_map;
class DisksApp : public Poco::Util::Application, public Loggers
class DisksApp : public Poco::Util::Application
{
public:
DisksApp() = default;
~DisksApp() override;
void addOptions();
void init(std::vector<String> & common_arguments);
int main(const std::vector<String> & args) override;
protected:
static String getDefaultConfigFileName();
void addOptions(
ProgramOptionsDescription & options_description,
boost::program_options::positional_options_description & positional_options_description);
void processOptions();
void printHelpMessage(ProgramOptionsDescription & command_option_description);
bool processQueryText(const String & text);
size_t findCommandPos(std::vector<String> & common_arguments);
void init(const std::vector<String> & common_arguments);
int main(const std::vector<String> & /*args*/) override;
CommandPtr getCommandByName(const String & command) const;
void initializeHistoryFile();
static void parseAndCheckOptions(
const std::vector<String> & arguments, const ProgramOptionsDescription & options_description, CommandLineOptions & options);
void printEntryHelpMessage() const;
void printAvailableCommandsHelpMessage() const;
void printCommandHelpMessage(String command_name) const;
void printCommandHelpMessage(CommandPtr command) const;
String getCommandLineWithAliases(CommandPtr command) const;
std::vector<String> getCompletions(const String & prefix) const;
std::vector<String> getEmptyCompletion(String command_name) const;
~DisksApp() override;
private:
void parseAndCheckOptions(
ProgramOptionsDescription & options_description,
boost::program_options::positional_options_description & positional_options_description,
std::vector<String> & arguments);
void runInteractive();
void runInteractiveReplxx();
void runInteractiveTestMode();
String getDefaultConfigFileName();
std::vector<String> getCommandsToComplete(const String & command_prefix) const;
// Fields responsible for the REPL work
String history_file;
LineReader::Suggest suggest;
static LineReader::Patterns query_extenders;
static LineReader::Patterns query_delimiters;
static String word_break_characters;
// General command line arguments parsing fields
protected:
ContextMutablePtr global_context;
SharedContextHolder shared_context;
String command_name;
std::vector<String> command_arguments;
std::unordered_set<String> supported_commands;
ContextMutablePtr global_context;
ProgramOptionsDescription options_description;
CommandLineOptions options;
std::unordered_map<String, CommandPtr> command_descriptions;
po::variables_map options;
};
std::optional<String> query;
const std::unordered_map<String, String> aliases
= {{"cp", "copy"},
{"mv", "move"},
{"ls", "list"},
{"list_disks", "list-disks"},
{"ln", "link"},
{"rm", "remove"},
{"cat", "read"},
{"r", "read"},
{"w", "write"},
{"create", "touch"},
{"delete", "remove"},
{"ls-disks", "list-disks"},
{"ls_disks", "list-disks"},
{"packed_io", "packed-io"},
{"change-dir", "cd"},
{"change_dir", "cd"},
{"switch_disk", "switch-disk"},
{"current", "current_disk_with_path"},
{"current_disk", "current_disk_with_path"},
{"current_path", "current_disk_with_path"},
{"cur", "current_disk_with_path"}};
std::set<String> multidisk_commands = {"copy", "packed-io", "switch-disk", "cd"};
std::unique_ptr<DisksClient> client{};
};
}

263
programs/disks/DisksClient.cpp Normal file
View File

@ -0,0 +1,263 @@
#include "DisksClient.h"
#include <Client/ClientBase.h>
#include <Client/ReplxxLineReader.h>
#include <Disks/registerDisks.h>
#include <Common/Config/ConfigProcessor.h>
#include <Formats/registerFormats.h>
namespace ErrorCodes
{
extern const int BAD_ARGUMENTS;
extern const int LOGICAL_ERROR;
};
namespace DB
{
DiskWithPath::DiskWithPath(DiskPtr disk_, std::optional<String> path_) : disk(disk_)
{
if (path_.has_value())
{
if (!fs::path{path_.value()}.is_absolute())
{
throw Exception(ErrorCodes::BAD_ARGUMENTS, "Initializing path {} is not absolute", path_.value());
}
path = path_.value();
}
else
{
path = String{"/"};
}
String relative_path = normalizePathAndGetAsRelative(path);
if (disk->isDirectory(relative_path) || (relative_path.empty() && (disk->isDirectory("/"))))
{
return;
}
throw Exception(
ErrorCodes::BAD_ARGUMENTS,
"Initializing path {} (normalized path: {}) at disk {} is not a directory",
path,
relative_path,
disk->getName());
}
std::vector<String> DiskWithPath::listAllFilesByPath(const String & any_path) const
{
if (isDirectory(any_path))
{
std::vector<String> file_names;
disk->listFiles(getRelativeFromRoot(any_path), file_names);
return file_names;
}
else
{
return {};
}
}
std::vector<String> DiskWithPath::getAllFilesByPattern(const String & pattern) const
{
auto [path_before, path_after] = [&]() -> std::pair<String, String>
{
auto slash_pos = pattern.find_last_of('/');
if (slash_pos >= pattern.size())
{
return {"", pattern};
}
else
{
return {pattern.substr(0, slash_pos + 1), pattern.substr(slash_pos + 1, pattern.size() - slash_pos - 1)};
}
}();
if (!isDirectory(path_before))
{
return {};
}
else
{
std::vector<String> file_names = listAllFilesByPath(path_before);
std::vector<String> answer;
for (const auto & file_name : file_names)
{
if (file_name.starts_with(path_after))
{
String file_pattern = path_before + file_name;
if (isDirectory(file_pattern))
{
file_pattern = file_pattern + "/";
}
answer.push_back(file_pattern);
}
}
return answer;
}
};
void DiskWithPath::setPath(const String & any_path)
{
if (isDirectory(any_path))
{
path = getAbsolutePath(any_path);
}
else
{
throw Exception(ErrorCodes::BAD_ARGUMENTS, "Path {} at disk {} is not a directory", any_path, disk->getName());
}
}
String DiskWithPath::validatePathAndGetAsRelative(const String & path)
{
String lexically_normal_path = fs::path(path).lexically_normal();
if (lexically_normal_path.find("..") != std::string::npos)
throw DB::Exception(DB::ErrorCodes::BAD_ARGUMENTS, "Path {} is not normalized", path);
/// If path is absolute we should keep it as relative inside disk, so disk will look like
/// an ordinary filesystem with root.
if (fs::path(lexically_normal_path).is_absolute())
return lexically_normal_path.substr(1);
return lexically_normal_path;
}
String DiskWithPath::normalizePathAndGetAsRelative(const String & messyPath)
{
std::filesystem::path path(messyPath);
std::filesystem::path canonical_path = std::filesystem::weakly_canonical(path);
String npath = canonical_path.make_preferred().string();
return validatePathAndGetAsRelative(npath);
}
String DiskWithPath::normalizePath(const String & path)
{
std::filesystem::path canonical_path = std::filesystem::weakly_canonical(path);
return canonical_path.make_preferred().string();
}
DisksClient::DisksClient(std::vector<std::pair<DiskPtr, std::optional<String>>> && disks_with_paths, std::optional<String> begin_disk)
{
if (disks_with_paths.empty())
{
throw Exception(ErrorCodes::BAD_ARGUMENTS, "Initializing array of disks is empty");
}
if (!begin_disk.has_value())
{
begin_disk = disks_with_paths[0].first->getName();
}
bool has_begin_disk = false;
for (auto & [disk, path] : disks_with_paths)
{
addDisk(disk, path);
if (disk->getName() == begin_disk.value())
{
has_begin_disk = true;
}
}
if (!has_begin_disk)
{
throw Exception(ErrorCodes::BAD_ARGUMENTS, "There is no begin_disk '{}' in initializing array", begin_disk.value());
}
current_disk = std::move(begin_disk.value());
}
const DiskWithPath & DisksClient::getDiskWithPath(const String & disk) const
{
try
{
return disks.at(disk);
}
catch (...)
{
throw Exception(ErrorCodes::BAD_ARGUMENTS, "The disk '{}' is unknown", disk);
}
}
DiskWithPath & DisksClient::getDiskWithPath(const String & disk)
{
try
{
return disks.at(disk);
}
catch (...)
{
throw Exception(ErrorCodes::BAD_ARGUMENTS, "The disk '{}' is unknown", disk);
}
}
const DiskWithPath & DisksClient::getCurrentDiskWithPath() const
{
try
{
return disks.at(current_disk);
}
catch (...)
{
throw Exception(ErrorCodes::LOGICAL_ERROR, "There is no current disk in client");
}
}
DiskWithPath & DisksClient::getCurrentDiskWithPath()
{
try
{
return disks.at(current_disk);
}
catch (...)
{
throw Exception(ErrorCodes::LOGICAL_ERROR, "There is no current disk in client");
}
}
void DisksClient::switchToDisk(const String & disk_, const std::optional<String> & path_)
{
if (disks.contains(disk_))
{
if (path_.has_value())
{
disks.at(disk_).setPath(path_.value());
}
current_disk = disk_;
}
else
{
throw Exception(ErrorCodes::BAD_ARGUMENTS, "The disk '{}' is unknown", disk_);
}
}
std::vector<String> DisksClient::getAllDiskNames() const
{
std::vector<String> answer{};
answer.reserve(disks.size());
for (const auto & [disk_name, _] : disks)
{
answer.push_back(disk_name);
}
return answer;
}
std::vector<String> DisksClient::getAllFilesByPatternFromAllDisks(const String & pattern) const
{
std::vector<String> answer{};
for (const auto & [_, disk] : disks)
{
for (auto & word : disk.getAllFilesByPattern(pattern))
{
answer.push_back(word);
}
}
return answer;
}
void DisksClient::addDisk(DiskPtr disk_, const std::optional<String> & path_)
{
String disk_name = disk_->getName();
if (disks.contains(disk_->getName()))
{
throw Exception(ErrorCodes::BAD_ARGUMENTS, "The disk '{}' already exists", disk_name);
}
disks.emplace(disk_name, DiskWithPath{disk_, path_});
}
}

89
programs/disks/DisksClient.h Normal file
View File

@ -0,0 +1,89 @@
#pragma once
#include <filesystem>
#include <optional>
#include <string>
#include <unordered_map>
#include <vector>
#include <Client/ReplxxLineReader.h>
#include <Loggers/Loggers.h>
#include "Disks/IDisk.h"
#include <Interpreters/Context.h>
#include <boost/program_options/options_description.hpp>
#include <boost/program_options/variables_map.hpp>
namespace fs = std::filesystem;
namespace DB
{
std::vector<String> split(const String & text, const String & delimiters);
using ProgramOptionsDescription = boost::program_options::options_description;
using CommandLineOptions = boost::program_options::variables_map;
class DiskWithPath
{
public:
explicit DiskWithPath(DiskPtr disk_, std::optional<String> path_ = std::nullopt);
String getAbsolutePath(const String & any_path) const { return normalizePath(fs::path(path) / any_path); }
String getCurrentPath() const { return path; }
bool isDirectory(const String & any_path) const
{
return disk->isDirectory(getRelativeFromRoot(any_path)) || (getRelativeFromRoot(any_path).empty() && (disk->isDirectory("/")));
}
std::vector<String> listAllFilesByPath(const String & any_path) const;
std::vector<String> getAllFilesByPattern(const String & pattern) const;
DiskPtr getDisk() const { return disk; }
void setPath(const String & any_path);
String getRelativeFromRoot(const String & any_path) const { return normalizePathAndGetAsRelative(getAbsolutePath(any_path)); }
private:
static String validatePathAndGetAsRelative(const String & path);
static std::string normalizePathAndGetAsRelative(const std::string & messyPath);
static std::string normalizePath(const std::string & messyPath);
const DiskPtr disk;
String path;
};
class DisksClient
{
public:
explicit DisksClient(std::vector<std::pair<DiskPtr, std::optional<String>>> && disks_with_paths, std::optional<String> begin_disk);
const DiskWithPath & getDiskWithPath(const String & disk) const;
DiskWithPath & getDiskWithPath(const String & disk);
const DiskWithPath & getCurrentDiskWithPath() const;
DiskWithPath & getCurrentDiskWithPath();
DiskPtr getCurrentDisk() const { return getCurrentDiskWithPath().getDisk(); }
DiskPtr getDisk(const String & disk) const { return getDiskWithPath(disk).getDisk(); }
void switchToDisk(const String & disk_, const std::optional<String> & path_);
std::vector<String> getAllDiskNames() const;
std::vector<String> getAllFilesByPatternFromAllDisks(const String & pattern) const;
private:
void addDisk(DiskPtr disk_, const std::optional<String> & path_);
String current_disk;
std::unordered_map<String, DiskWithPath> disks;
};
}

67
programs/disks/ICommand.cpp
View File

@ -1,5 +1,5 @@
#include "ICommand.h"
#include <iostream>
#include "DisksClient.h"
namespace DB
@ -10,43 +10,42 @@ namespace ErrorCodes
extern const int BAD_ARGUMENTS;
}
void ICommand::printHelpMessage() const
CommandLineOptions ICommand::processCommandLineArguments(const Strings & commands)
{
std::cout << "Command: " << command_name << '\n';
std::cout << "Description: " << description << '\n';
std::cout << "Usage: " << usage << '\n';
CommandLineOptions options;
auto parser = po::command_line_parser(commands);
parser.options(options_description).positional(positional_options_description);
if (command_option_description)
po::parsed_options parsed = parser.run();
po::store(parsed, options);
return options;
}
void ICommand::execute(const Strings & commands, DisksClient & client)
{
try
{
auto options = *command_option_description;
if (!options.options().empty())
std::cout << options << '\n';
processCommandLineArguments(commands);
}
catch (std::exception & exc)
{
throw Exception(ErrorCodes::BAD_ARGUMENTS, "{}", exc.what());
}
executeImpl(processCommandLineArguments(commands), client);
}
DiskWithPath & ICommand::getDiskWithPath(DisksClient & client, const CommandLineOptions & options, const String & name)
{
auto disk_name = getValueFromCommandLineOptionsWithOptional<String>(options, name);
if (disk_name.has_value())
{
return client.getDiskWithPath(disk_name.value());
}
else
{
return client.getCurrentDiskWithPath();
}
}
void ICommand::addOptions(ProgramOptionsDescription & options_description)
{
if (!command_option_description || command_option_description->options().empty())
return;
options_description.add(*command_option_description);
}
String ICommand::validatePathAndGetAsRelative(const String & path)
{
/// If path contain non-normalized symbols like . we will normalized them. If the resulting normalized path
/// still contain '..' it can be dangerous, disallow such paths. Also since clickhouse-disks
/// is not an interactive program (don't track you current path) it's OK to disallow .. paths.
String lexically_normal_path = fs::path(path).lexically_normal();
if (lexically_normal_path.find("..") != std::string::npos)
throw DB::Exception(DB::ErrorCodes::BAD_ARGUMENTS, "Path {} is not normalized", path);
/// If path is absolute we should keep it as relative inside disk, so disk will look like
/// an ordinary filesystem with root.
if (fs::path(lexically_normal_path).is_absolute())
return lexically_normal_path.substr(1);
return lexically_normal_path;
}
}

130
programs/disks/ICommand.h
View File

@ -1,66 +1,146 @@
#pragma once
#include <Disks/IDisk.h>
#include <optional>
#include <Disks/DiskSelector.h>
#include <Disks/IDisk.h>
#include <boost/any/bad_any_cast.hpp>
#include <boost/program_options.hpp>
#include <Common/Config/ConfigProcessor.h>
#include <Poco/Util/Application.h>
#include "Common/Exception.h"
#include <Common/Config/ConfigProcessor.h>
#include <memory>
#include <boost/program_options/positional_options.hpp>
#include "DisksApp.h"
#include "DisksClient.h"
#include "ICommand_fwd.h"
namespace DB
{
namespace po = boost::program_options;
using ProgramOptionsDescription = boost::program_options::options_description;
using CommandLineOptions = boost::program_options::variables_map;
using ProgramOptionsDescription = po::options_description;
using PositionalProgramOptionsDescription = po::positional_options_description;
using CommandLineOptions = po::variables_map;
namespace ErrorCodes
{
extern const int BAD_ARGUMENTS;
}
class ICommand
{
public:
ICommand() = default;
explicit ICommand() = default;
virtual ~ICommand() = default;
virtual void execute(
const std::vector<String> & command_arguments,
std::shared_ptr<DiskSelector> & disk_selector,
Poco::Util::LayeredConfiguration & config) = 0;
void execute(const Strings & commands, DisksClient & client);
const std::optional<ProgramOptionsDescription> & getCommandOptions() const { return command_option_description; }
virtual void executeImpl(const CommandLineOptions & options, DisksClient & client) = 0;
void addOptions(ProgramOptionsDescription & options_description);
virtual void processOptions(Poco::Util::LayeredConfiguration & config, po::variables_map & options) const = 0;
CommandLineOptions processCommandLineArguments(const Strings & commands);
protected:
void printHelpMessage() const;
template <typename T>
static T getValueFromCommandLineOptions(const CommandLineOptions & options, const String & name)
{
try
{
return options[name].as<T>();
}
catch (boost::bad_any_cast &)
{
throw DB::Exception(ErrorCodes::BAD_ARGUMENTS, "Argument '{}' has wrong type and can't be parsed", name);
}
}
template <typename T>
static T getValueFromCommandLineOptionsThrow(const CommandLineOptions & options, const String & name)
{
if (options.count(name))
{
return getValueFromCommandLineOptions<T>(options, name);
}
else
{
throw DB::Exception(ErrorCodes::BAD_ARGUMENTS, "Mandatory argument '{}' is missing", name);
}
}
template <typename T>
static T getValueFromCommandLineOptionsWithDefault(const CommandLineOptions & options, const String & name, const T & default_value)
{
if (options.count(name))
{
return getValueFromCommandLineOptions<T>(options, name);
}
else
{
return default_value;
}
}
template <typename T>
static std::optional<T> getValueFromCommandLineOptionsWithOptional(const CommandLineOptions & options, const String & name)
{
if (options.count(name))
{
return std::optional{getValueFromCommandLineOptions<T>(options, name)};
}
else
{
return std::nullopt;
}
}
DiskWithPath & getDiskWithPath(DisksClient & client, const CommandLineOptions & options, const String & name);
String getTargetLocation(const String & path_from, DiskWithPath & disk_to, const String & path_to)
{
if (!disk_to.getDisk()->isDirectory(path_to))
{
return path_to;
}
String copied_path_from = path_from;
if (copied_path_from.ends_with('/'))
{
copied_path_from.pop_back();
}
String plain_filename = fs::path(copied_path_from).filename();
return fs::path{path_to} / plain_filename;
}
static String validatePathAndGetAsRelative(const String & path);
public:
String command_name;
String description;
ProgramOptionsDescription options_description;
protected:
std::optional<ProgramOptionsDescription> command_option_description;
String usage;
po::positional_options_description positional_options_description;
PositionalProgramOptionsDescription positional_options_description;
};
using CommandPtr = std::unique_ptr<ICommand>;
}
DB::CommandPtr makeCommandCopy();
DB::CommandPtr makeCommandLink();
DB::CommandPtr makeCommandList();
DB::CommandPtr makeCommandListDisks();
DB::CommandPtr makeCommandList();
DB::CommandPtr makeCommandChangeDirectory();
DB::CommandPtr makeCommandLink();
DB::CommandPtr makeCommandMove();
DB::CommandPtr makeCommandRead();
DB::CommandPtr makeCommandRemove();
DB::CommandPtr makeCommandWrite();
DB::CommandPtr makeCommandMkDir();
DB::CommandPtr makeCommandSwitchDisk();
DB::CommandPtr makeCommandGetCurrentDiskAndPath();
DB::CommandPtr makeCommandHelp(const DisksApp & disks_app);
DB::CommandPtr makeCommandTouch();
#ifdef CLICKHOUSE_CLOUD
DB::CommandPtr makeCommandPackedIO();
#endif
}

10
programs/disks/ICommand_fwd.h Normal file
View File

@ -0,0 +1,10 @@
#pragma once
#include <memory>
namespace DB
{
class ICommand;
using CommandPtr = std::shared_ptr<ICommand>;
}

8
src/Disks/DiskFactory.cpp
View File

@ -27,7 +27,8 @@ DiskPtr DiskFactory::create(
ContextPtr context,
const DisksMap & map,
bool attach,
bool custom_disk) const
bool custom_disk,
const std::unordered_set<String> & skip_types) const
{
const auto disk_type = config.getString(config_prefix + ".type", "local");
@ -38,6 +39,11 @@ DiskPtr DiskFactory::create(
"DiskFactory: the disk '{}' has unknown disk type: {}", name, disk_type);
}
if (skip_types.contains(found->first))
{
return nullptr;
}
const auto & disk_creator = found->second;
return disk_creator(name, config, config_prefix, context, map, attach, custom_disk);
}

3
src/Disks/DiskFactory.h
View File

@ -42,7 +42,8 @@ public:
ContextPtr context,
const DisksMap & map,
bool attach = false,
bool custom_disk = false) const;
bool custom_disk = false,
const std::unordered_set<String> & skip_types = {}) const;
private:
using DiskTypeRegistry = std::unordered_map<String, Creator>;

38
src/Disks/DiskSelector.cpp
View File

@ -7,7 +7,6 @@
#include <Common/logger_useful.h>
#include <Interpreters/Context.h>
#include <set>
namespace DB
{
@ -27,7 +26,8 @@ void DiskSelector::assertInitialized() const
}
void DiskSelector::initialize(const Poco::Util::AbstractConfiguration & config, const String & config_prefix, ContextPtr context, DiskValidator disk_validator)
void DiskSelector::initialize(
const Poco::Util::AbstractConfiguration & config, const String & config_prefix, ContextPtr context, DiskValidator disk_validator)
{
Poco::Util::AbstractConfiguration::Keys keys;
config.keys(config_prefix, keys);
@ -36,6 +36,8 @@ void DiskSelector::initialize(const Poco::Util::AbstractConfiguration & config,
constexpr auto default_disk_name = "default";
bool has_default_disk = false;
constexpr auto local_disk_name = "local";
bool has_local_disk = false;
for (const auto & disk_name : keys)
{
if (!std::all_of(disk_name.begin(), disk_name.end(), isWordCharASCII))
@ -44,21 +46,31 @@ void DiskSelector::initialize(const Poco::Util::AbstractConfiguration & config,
if (disk_name == default_disk_name)
has_default_disk = true;
if (disk_name == local_disk_name)
has_local_disk = true;
const auto disk_config_prefix = config_prefix + "." + disk_name;
if (disk_validator && !disk_validator(config, disk_config_prefix, disk_name))
continue;
disks.emplace(disk_name, factory.create(disk_name, config, disk_config_prefix, context, disks));
auto created_disk
= factory.create(disk_name, config, disk_config_prefix, context, disks, /*attach*/ false, /*custom_disk*/ false, skip_types);
if (created_disk.get())
{
disks.emplace(disk_name, std::move(created_disk));
}
}
if (!has_default_disk)
{
disks.emplace(
default_disk_name,
std::make_shared<DiskLocal>(
default_disk_name, context->getPath(), 0, context, config, config_prefix));
default_disk_name, std::make_shared<DiskLocal>(default_disk_name, context->getPath(), 0, context, config, config_prefix));
}
if (!has_local_disk && (context->getApplicationType() == Context::ApplicationType::DISKS))
{
throw_away_local_on_update = true;
disks.emplace(local_disk_name, std::make_shared<DiskLocal>(local_disk_name, "/", 0, context, config, config_prefix));
}
is_initialized = true;
}
@ -76,6 +88,7 @@ DiskSelectorPtr DiskSelector::updateFromConfig(
std::shared_ptr<DiskSelector> result = std::make_shared<DiskSelector>(*this);
constexpr auto default_disk_name = "default";
constexpr auto local_disk_name = "local";
DisksMap old_disks_minus_new_disks(result->getDisksMap());
for (const auto & disk_name : keys)
@ -86,7 +99,12 @@ DiskSelectorPtr DiskSelector::updateFromConfig(
auto disk_config_prefix = config_prefix + "." + disk_name;
if (!result->getDisksMap().contains(disk_name))
{
result->addToDiskMap(disk_name, factory.create(disk_name, config, disk_config_prefix, context, result->getDisksMap()));
auto created_disk = factory.create(
disk_name, config, disk_config_prefix, context, result->getDisksMap(), /*attach*/ false, /*custom_disk*/ false, skip_types);
if (created_disk)
{
result->addToDiskMap(disk_name, created_disk);
}
}
else
{
@ -99,6 +117,10 @@ DiskSelectorPtr DiskSelector::updateFromConfig(
}
old_disks_minus_new_disks.erase(default_disk_name);
if (throw_away_local_on_update)
{
old_disks_minus_new_disks.erase(local_disk_name);
}
if (!old_disks_minus_new_disks.empty())
{

6
src/Disks/DiskSelector.h
View File

@ -20,7 +20,7 @@ class DiskSelector
public:
static constexpr auto TMP_INTERNAL_DISK_PREFIX = "__tmp_internal_";
DiskSelector() = default;
explicit DiskSelector(std::unordered_set<String> skip_types_ = {}) : skip_types(skip_types_) { }
DiskSelector(const DiskSelector & from) = default;
using DiskValidator = std::function<bool(const Poco::Util::AbstractConfiguration & config, const String & disk_config_prefix, const String & disk_name)>;
@ -48,6 +48,10 @@ private:
bool is_initialized = false;
void assertInitialized() const;
const std::unordered_set<String> skip_types;
bool throw_away_local_on_update = false;
};
}

165
tests/integration/test_disks_app_func/test.py
View File

@ -9,7 +9,9 @@ def started_cluster():
try:
cluster = ClickHouseCluster(__file__)
cluster.add_instance(
"disks_app_test", main_configs=["config.xml"], with_minio=True
"disks_app_test",
main_configs=["config.xml"],
with_minio=True,
)
cluster.start()
@ -47,12 +49,18 @@ def test_disks_app_func_ld(started_cluster):
source = cluster.instances["disks_app_test"]
out = source.exec_in_container(
["/usr/bin/clickhouse", "disks", "--save-logs", "list-disks"]
["/usr/bin/clickhouse", "disks", "--save-logs", "--query", "list-disks"]
)
disks = out.split("\n")
disks = list(
sorted(
map(
lambda x: x.split(":")[0], filter(lambda x: len(x) > 1, out.split("\n"))
)
)
)
assert disks[0] == "default" and disks[1] == "test1" and disks[2] == "test2"
assert disks[:4] == ["default", "local", "test1", "test2"]
def test_disks_app_func_ls(started_cluster):
@ -61,7 +69,15 @@ def test_disks_app_func_ls(started_cluster):
init_data(source)
out = source.exec_in_container(
["/usr/bin/clickhouse", "disks", "--save-logs", "--disk", "test1", "list", "."]
[
"/usr/bin/clickhouse",
"disks",
"--save-logs",
"--disk",
"test1",
"--query",
"list .",
]
)
files = out.split("\n")
@ -75,9 +91,8 @@ def test_disks_app_func_ls(started_cluster):
"--save-logs",
"--disk",
"test1",
"list",
".",
"--recursive",
"--query",
"list . --recursive",
]
)
@ -102,8 +117,8 @@ def test_disks_app_func_cp(started_cluster):
"--save-logs",
"--disk",
"test1",
"write",
"path1",
"--query",
"'write path1'",
]
),
]
@ -113,18 +128,21 @@ def test_disks_app_func_cp(started_cluster):
[
"/usr/bin/clickhouse",
"disks",
"copy",
"--disk-from",
"test1",
"--disk-to",
"test2",
".",
".",
"--query",
"copy --recursive --disk-from test1 --disk-to test2 . .",
]
)
out = source.exec_in_container(
["/usr/bin/clickhouse", "disks", "--save-logs", "--disk", "test2", "list", "."]
[
"/usr/bin/clickhouse",
"disks",
"--save-logs",
"--disk",
"test2",
"--query",
"list .",
]
)
assert "path1" in out
@ -136,8 +154,8 @@ def test_disks_app_func_cp(started_cluster):
"--save-logs",
"--disk",
"test2",
"remove",
"path1",
"--query",
"remove path1",
]
)
@ -148,21 +166,37 @@ def test_disks_app_func_cp(started_cluster):
"--save-logs",
"--disk",
"test1",
"remove",
"path1",
"--query",
"remove path1",
]
)
# alesapin: Why we need list one more time?
# kssenii: it is an assertion that the file is indeed deleted
out = source.exec_in_container(
["/usr/bin/clickhouse", "disks", "--save-logs", "--disk", "test2", "list", "."]
[
"/usr/bin/clickhouse",
"disks",
"--save-logs",
"--disk",
"test2",
"--query",
"list .",
]
)
assert "path1" not in out
out = source.exec_in_container(
["/usr/bin/clickhouse", "disks", "--save-logs", "--disk", "test1", "list", "."]
[
"/usr/bin/clickhouse",
"disks",
"--save-logs",
"--disk",
"test1",
"--query",
"list .",
]
)
assert "path1" not in out
@ -177,14 +211,13 @@ def test_disks_app_func_ln(started_cluster):
[
"/usr/bin/clickhouse",
"disks",
"link",
"data/default/test_table",
"data/default/z_tester",
"--query",
"link data/default/test_table data/default/z_tester",
]
)
out = source.exec_in_container(
["/usr/bin/clickhouse", "disks", "--save-logs", "list", "data/default/"]
["/usr/bin/clickhouse", "disks", "--save-logs", "--query", "list data/default/"]
)
files = out.split("\n")
@ -209,15 +242,23 @@ def test_disks_app_func_rm(started_cluster):
"--save-logs",
"--disk",
"test2",
"write",
"path3",
"--query",
"'write path3'",
]
),
]
)
out = source.exec_in_container(
["/usr/bin/clickhouse", "disks", "--save-logs", "--disk", "test2", "list", "."]
[
"/usr/bin/clickhouse",
"disks",
"--save-logs",
"--disk",
"test2",
"--query",
"list .",
]
)
assert "path3" in out
@ -229,13 +270,21 @@ def test_disks_app_func_rm(started_cluster):
"--save-logs",
"--disk",
"test2",
"remove",
"path3",
"--query",
"remove path3",
]
)
out = source.exec_in_container(
["/usr/bin/clickhouse", "disks", "--save-logs", "--disk", "test2", "list", "."]
[
"/usr/bin/clickhouse",
"disks",
"--save-logs",
"--disk",
"test2",
"--query",
"list .",
]
)
assert "path3" not in out
@ -247,7 +296,15 @@ def test_disks_app_func_mv(started_cluster):
init_data(source)
out = source.exec_in_container(
["/usr/bin/clickhouse", "disks", "--save-logs", "--disk", "test1", "list", "."]
[
"/usr/bin/clickhouse",
"disks",
"--save-logs",
"--disk",
"test1",
"--query",
"list .",
]
)
files = out.split("\n")
@ -260,14 +317,21 @@ def test_disks_app_func_mv(started_cluster):
"disks",
"--disk",
"test1",
"move",
"store",
"old_store",
"--query",
"move store old_store",
]
)
out = source.exec_in_container(
["/usr/bin/clickhouse", "disks", "--save-logs", "--disk", "test1", "list", "."]
[
"/usr/bin/clickhouse",
"disks",
"--save-logs",
"--disk",
"test1",
"--query",
"list .",
]
)
files = out.split("\n")
@ -290,8 +354,8 @@ def test_disks_app_func_read_write(started_cluster):
"--save-logs",
"--disk",
"test1",
"write",
"5.txt",
"--query",
"'write 5.txt'",
]
),
]
@ -304,8 +368,8 @@ def test_disks_app_func_read_write(started_cluster):
"--save-logs",
"--disk",
"test1",
"read",
"5.txt",
"--query",
"read 5.txt",
]
)
@ -319,7 +383,15 @@ def test_remote_disk_list(started_cluster):
init_data_s3(source)
out = source.exec_in_container(
["/usr/bin/clickhouse", "disks", "--save-logs", "--disk", "test3", "list", "."]
[
"/usr/bin/clickhouse",
"disks",
"--save-logs",
"--disk",
"test3",
"--query",
"list .",
]
)
files = out.split("\n")
@ -333,9 +405,8 @@ def test_remote_disk_list(started_cluster):
"--save-logs",
"--disk",
"test3",
"list",
".",
"--recursive",
"--query",
"list . --recursive",
]
)

0
tests/integration/test_disks_app_interactive/__init__.py Normal file
View File

3
tests/integration/test_disks_app_interactive/configs/config.xml Normal file
View File

@ -0,0 +1,3 @@
<clickhouse>
<path>/var/lib/clickhouse/</path>
</clickhouse>

331
tests/integration/test_disks_app_interactive/test.py Normal file
View File

@ -0,0 +1,331 @@
from helpers.cluster import ClickHouseCluster
import pytest
import pathlib
import subprocess
import select
import io
from typing import List, Tuple, Dict, Union, Optional
import os
class ClickHouseDisksException(Exception):
pass
@pytest.fixture(scope="module")
def started_cluster():
global cluster
try:
cluster = ClickHouseCluster(__file__)
cluster.add_instance(
"disks_app_test",
main_configs=["server_configs/config.xml"],
with_minio=True,
)
cluster.start()
yield cluster
finally:
cluster.shutdown()
class DisksClient(object):
SEPARATOR = b"\a\a\a\a\n"
local_client: Optional["DisksClient"] = None # static variable
default_disk_root_directory: str = "/var/lib/clickhouse"
def __init__(self, bin_path: str, config_path: str, working_path: str):
self.bin_path = bin_path
self.working_path = working_path
self.proc = subprocess.Popen(
[bin_path, "disks", "--test-mode", "--config", config_path],
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
)
self.poller = select.epoll()
self.poller.register(self.proc.stdout)
self.poller.register(self.proc.stderr)
self.stopped = False
self._fd_nums = {
self.proc.stdout.fileno(): self.proc.stdout,
self.proc.stderr.fileno(): self.proc.stderr,
}
def execute_query(self, query: str, timeout: float = 5.0) -> str:
output = io.BytesIO()
self.proc.stdin.write(query.encode() + b"\n")
self.proc.stdin.flush()
events = self.poller.poll(timeout)
if not events:
raise TimeoutError(f"Disks client returned no output")
for fd_num, event in events:
if event & (select.EPOLLIN | select.EPOLLPRI):
file = self._fd_nums[fd_num]
if file == self.proc.stdout:
while True:
chunk = file.readline()
if chunk.endswith(self.SEPARATOR):
break
output.write(chunk)
elif file == self.proc.stderr:
error_line = self.proc.stderr.readline()
print(error_line)
raise ClickHouseDisksException(error_line.strip().decode())
else:
raise ValueError(f"Failed to read from pipe. Flag {event}")
data = output.getvalue().strip().decode()
return data
def list_disks(self) -> List[Tuple[str, str]]:
output = self.execute_query("list-disks")
return list(
sorted(
map(
lambda x: (x.split(":")[0], ":".join(x.split(":")[1:])),
output.split("\n"),
)
)
)
def current_disk_with_path(self) -> Tuple[str, str]:
output = self.execute_query("current_disk_with_path")
disk_line = output.split("\n")[0]
path_line = output.split("\n")[1]
assert disk_line.startswith("Disk: ")
assert path_line.startswith("Path: ")
return disk_line[6:], path_line[6:]
def ls(
self, path: str, recursive: bool = False, show_hidden: bool = False
) -> Union[List[str], Dict[str, List[str]]]:
recursive_adding = "--recursive " if recursive else ""
show_hidden_adding = "--all " if show_hidden else ""
output = self.execute_query(
f"list {path} {recursive_adding} {show_hidden_adding}"
)
if recursive:
answer: Dict[str, List[str]] = dict()
blocks = output.split("\n\n")
for block in blocks:
directory = block.split("\n")[0][:-1]
files = block.split("\n")[1:]
answer[directory] = files
return answer
else:
return output.split("\n")
def switch_disk(self, disk: str, directory: Optional[str] = None):
directory_addition = f"--path {directory} " if directory is not None else ""
self.execute_query(f"switch-disk {disk} {directory_addition}")
def cd(self, directory: str, disk: Optional[str] = None):
disk_addition = f"--disk {disk} " if disk is not None else ""
self.execute_query(f"cd {directory} {disk_addition}")
def copy(
self,
path_from,
path_to,
disk_from: Optional[str] = None,
disk_to: Optional[str] = None,
recursive: bool = False,
):
disk_from_option = f"--disk-from {disk_from} " if disk_from is not None else ""
disk_to_option = f"--disk-to {disk_to} " if disk_to is not None else ""
recursive_tag = "--recursive" if recursive else ""
self.execute_query(
f"copy {recursive_tag} {path_from} {path_to} {disk_from_option} {disk_to_option}"
)
def move(self, path_from: str, path_to: str):
self.execute_query(f"move {path_from} {path_to}")
def rm(self, path: str, recursive: bool = False):
recursive_tag = "--recursive" if recursive else ""
self.execute_query(f"rm {recursive_tag} {path}")
def mkdir(self, path: str, recursive: bool = False):
recursive_adding = "--recursive " if recursive else ""
self.execute_query(f"mkdir {path} {recursive_adding}")
def ln(self, path_from: str, path_to: str):
self.execute_query(f"link {path_from} {path_to}")
def read(self, path_from: str, path_to: Optional[str] = None):
path_to_adding = f"--path-to {path_to} " if path_to is not None else ""
output = self.execute_query(f"read {path_from} {path_to_adding}")
return output
def write(
self, path_from: str, path_to: str
): # Writing from stdin is difficult to test (do not know how to do this in python)
path_from_adding = f"--path-from {path_from}"
self.execute_query(f"write {path_from_adding} {path_to}")
@staticmethod
def getLocalDisksClient(refresh: bool):
if (DisksClient.local_client is None) or refresh:
binary_file = os.environ.get("CLICKHOUSE_TESTS_SERVER_BIN_PATH")
current_working_directory = str(pathlib.Path().resolve())
config_file = f"{current_working_directory}/test_disks_app_interactive/configs/config.xml"
if not os.path.exists(DisksClient.default_disk_root_directory):
os.mkdir(DisksClient.default_disk_root_directory)
DisksClient.local_client = DisksClient(
binary_file, config_file, current_working_directory
)
return DisksClient.local_client
else:
return DisksClient.local_client
def test_disks_app_interactive_list_disks():
client = DisksClient.getLocalDisksClient(True)
expected_disks_with_path = [
("default", "/"),
("local", client.working_path),
]
assert expected_disks_with_path == client.list_disks()
assert client.current_disk_with_path() == ("default", "/")
client.switch_disk("local")
assert client.current_disk_with_path() == (
"local",
client.working_path,
)
def test_disks_app_interactive_list_files_local():
client = DisksClient.getLocalDisksClient(True)
client.switch_disk("local")
excepted_listed_files = sorted(os.listdir("test_disks_app_interactive/"))
listed_files = sorted(client.ls("test_disks_app_interactive/"))
assert excepted_listed_files == listed_files
def test_disks_app_interactive_list_directories_default():
client = DisksClient.getLocalDisksClient(True)
traversed_dir = client.ls(".", recursive=True)
client.mkdir("dir1")
client.mkdir("dir2")
client.mkdir(".dir3")
client.cd("dir1")
client.mkdir("dir11")
client.mkdir(".dir12")
client.mkdir("dir13")
client.cd("../dir2")
client.mkdir("dir21")
client.mkdir("dir22")
client.mkdir(".dir23")
client.cd("../.dir3")
client.mkdir("dir31")
client.mkdir(".dir32")
client.cd("..")
traversed_dir = client.ls(".", recursive=True)
assert traversed_dir == {
".": ["dir1", "dir2"],
"./dir1": ["dir11", "dir13"],
"./dir2": ["dir21", "dir22"],
"./dir1/dir11": [],
"./dir1/dir13": [],
"./dir2/dir21": [],
"./dir2/dir22": [],
}
traversed_dir = client.ls(".", recursive=True, show_hidden=True)
assert traversed_dir == {
".": [".dir3", "dir1", "dir2"],
"./dir1": [".dir12", "dir11", "dir13"],
"./dir2": [".dir23", "dir21", "dir22"],
"./.dir3": [".dir32", "dir31"],
"./dir1/dir11": [],
"./dir1/.dir12": [],
"./dir1/dir13": [],
"./dir2/dir21": [],
"./dir2/dir22": [],
"./dir2/.dir23": [],
"./.dir3/dir31": [],
"./.dir3/.dir32": [],
}
client.rm("dir2", recursive=True)
traversed_dir = client.ls(".", recursive=True, show_hidden=True)
assert traversed_dir == {
".": [".dir3", "dir1"],
"./dir1": [".dir12", "dir11", "dir13"],
"./.dir3": [".dir32", "dir31"],
"./dir1/dir11": [],
"./dir1/.dir12": [],
"./dir1/dir13": [],
"./.dir3/dir31": [],
"./.dir3/.dir32": [],
}
traversed_dir = client.ls(".", recursive=True, show_hidden=False)
assert traversed_dir == {
".": ["dir1"],
"./dir1": ["dir11", "dir13"],
"./dir1/dir11": [],
"./dir1/dir13": [],
}
client.rm("dir1", recursive=True)
client.rm(".dir3", recursive=True)
assert client.ls(".", recursive=True, show_hidden=False) == {".": []}
def test_disks_app_interactive_cp_and_read():
initial_text = "File content"
with open("a.txt", "w") as file:
file.write(initial_text)
client = DisksClient.getLocalDisksClient(True)
client.switch_disk("default")
client.copy("a.txt", "/a.txt", disk_from="local", disk_to="default")
read_text = client.read("a.txt")
assert initial_text == read_text
client.mkdir("dir1")
client.copy("a.txt", "/dir1/b.txt", disk_from="local", disk_to="default")
read_text = client.read("a.txt", path_to="dir1/b.txt")
assert "" == read_text
read_text = client.read("/dir1/b.txt")
assert read_text == initial_text
with open(f"{DisksClient.default_disk_root_directory}/dir1/b.txt", "r") as file:
read_text = file.read()
assert read_text == initial_text
os.remove("a.txt")
client.rm("a.txt")
client.rm("/dir1", recursive=True)
def test_disks_app_interactive_test_move_and_write():
initial_text = "File content"
with open("a.txt", "w") as file:
file.write(initial_text)
client = DisksClient.getLocalDisksClient(True)
client.switch_disk("default")
client.copy("a.txt", "/a.txt", disk_from="local", disk_to="default")
files = client.ls(".")
assert files == ["a.txt"]
client.move("a.txt", "b.txt")
files = client.ls(".")
assert files == ["b.txt"]
read_text = client.read("/b.txt")
assert read_text == initial_text
client.write("b.txt", "c.txt")
read_text = client.read("c.txt")
assert read_text == initial_text
os.remove("a.txt")

16
tests/queries/0_stateless/02802_clickhouse_disks_s3_copy.sh
View File

@ -14,14 +14,11 @@ function run_test_for_disk()
echo "$disk"
clickhouse-disks -C "$config" --disk "$disk" write --input "$config" $CLICKHOUSE_DATABASE/test
clickhouse-disks -C "$config" --log-level test --disk "$disk" copy $CLICKHOUSE_DATABASE/test $CLICKHOUSE_DATABASE/test.copy |& {
clickhouse-disks -C "$config" --disk "$disk" --query "write --path-from $config $CLICKHOUSE_DATABASE/test"
clickhouse-disks -C "$config" --log-level test --disk "$disk" --query "copy -r $CLICKHOUSE_DATABASE/test $CLICKHOUSE_DATABASE/test.copy" |& {
grep -o -e "Single part upload has completed." -e "Single operation copy has completed."
}
clickhouse-disks -C "$config" --disk "$disk" remove $CLICKHOUSE_DATABASE/test
# NOTE: this is due to "copy" does works like "cp -R from to/" instead of "cp from to"
clickhouse-disks -C "$config" --disk "$disk" remove $CLICKHOUSE_DATABASE/test.copy/test
clickhouse-disks -C "$config" --disk "$disk" remove $CLICKHOUSE_DATABASE/test.copy
clickhouse-disks -C "$config" --disk "$disk" --query "remove -r $CLICKHOUSE_DATABASE/test"
}
function run_test_copy_from_s3_to_s3(){
@ -29,13 +26,12 @@ function run_test_copy_from_s3_to_s3(){
local disk_dest=$1 && shift
echo "copy from $disk_src to $disk_dest"
clickhouse-disks -C "$config" --disk "$disk_src" write --input "$config" $CLICKHOUSE_DATABASE/test
clickhouse-disks -C "$config" --disk "$disk_src" --query "write --path-from $config $CLICKHOUSE_DATABASE/test"
clickhouse-disks -C "$config" --log-level test copy --disk-from "$disk_src" --disk-to "$disk_dest" $CLICKHOUSE_DATABASE/test $CLICKHOUSE_DATABASE/test.copy |& {
clickhouse-disks -C "$config" --log-level test --query "copy -r --disk-from $disk_src --disk-to $disk_dest $CLICKHOUSE_DATABASE/test $CLICKHOUSE_DATABASE/test.copy" |& {
grep -o -e "Single part upload has completed." -e "Single operation copy has completed."
}
clickhouse-disks -C "$config" --disk "$disk_dest" remove $CLICKHOUSE_DATABASE/test.copy/test
clickhouse-disks -C "$config" --disk "$disk_dest" remove $CLICKHOUSE_DATABASE/test.copy
clickhouse-disks -C "$config" --disk "$disk_dest" --query "remove -r $CLICKHOUSE_DATABASE/test.copy"
}
run_test_for_disk s3_plain_native_copy

32
tests/queries/0_stateless/02980_s3_plain_DROP_TABLE_MergeTree.reference
View File

@ -3,28 +3,28 @@ data after ATTACH 1
Files before DETACH TABLE
all_1_1_0
backups/ordinary_default/data/ordinary_default/data/all_1_1_0:
primary.cidx
serialization.json
metadata_version.txt
default_compression_codec.txt
/backups/ordinary_default/data/ordinary_default/data/all_1_1_0:
checksums.txt
columns.txt
count.txt
data.bin
data.cmrk3
count.txt
columns.txt
checksums.txt
default_compression_codec.txt
metadata_version.txt
primary.cidx
serialization.json
Files after DETACH TABLE
all_1_1_0
backups/ordinary_default/data/ordinary_default/data/all_1_1_0:
primary.cidx
serialization.json
metadata_version.txt
default_compression_codec.txt
/backups/ordinary_default/data/ordinary_default/data/all_1_1_0:
checksums.txt
columns.txt
count.txt
data.bin
data.cmrk3
count.txt
columns.txt
checksums.txt
default_compression_codec.txt
metadata_version.txt
primary.cidx
serialization.json

4
tests/queries/0_stateless/02980_s3_plain_DROP_TABLE_MergeTree.sh
View File

@ -49,11 +49,11 @@ path=$($CLICKHOUSE_CLIENT -q "SELECT replace(data_paths[1], 's3_plain', '') FROM
path=${path%/}
echo "Files before DETACH TABLE"
clickhouse-disks -C "$config" --disk s3_plain_disk list --recursive "${path:?}" | tail -n+2
clickhouse-disks -C "$config" --disk s3_plain_disk --query "list --recursive $path" | tail -n+2
$CLICKHOUSE_CLIENT -q "detach table data"
echo "Files after DETACH TABLE"
clickhouse-disks -C "$config" --disk s3_plain_disk list --recursive "$path" | tail -n+2
clickhouse-disks -C "$config" --disk s3_plain_disk --query "list --recursive $path" | tail -n+2
# metadata file is left
$CLICKHOUSE_CLIENT --force_remove_data_recursively_on_drop=1 -q "drop database if exists $CLICKHOUSE_DATABASE"

32
tests/queries/0_stateless/02980_s3_plain_DROP_TABLE_ReplicatedMergeTree.reference
View File

@ -3,28 +3,28 @@ data after ATTACH 1
Files before DETACH TABLE
all_X_X_X
backups/ordinary_default/data/ordinary_default/data_read/all_X_X_X:
primary.cidx
serialization.json
metadata_version.txt
default_compression_codec.txt
/backups/ordinary_default/data/ordinary_default/data_read/all_X_X_X:
checksums.txt
columns.txt
count.txt
data.bin
data.cmrk3
count.txt
columns.txt
checksums.txt
default_compression_codec.txt
metadata_version.txt
primary.cidx
serialization.json
Files after DETACH TABLE
all_X_X_X
backups/ordinary_default/data/ordinary_default/data_read/all_X_X_X:
primary.cidx
serialization.json
metadata_version.txt
default_compression_codec.txt
/backups/ordinary_default/data/ordinary_default/data_read/all_X_X_X:
checksums.txt
columns.txt
count.txt
data.bin
data.cmrk3
count.txt
columns.txt
checksums.txt
default_compression_codec.txt
metadata_version.txt
primary.cidx
serialization.json

4
tests/queries/0_stateless/02980_s3_plain_DROP_TABLE_ReplicatedMergeTree.sh
View File

@ -55,14 +55,14 @@ path=${path%/}
echo "Files before DETACH TABLE"
# sed to match any part, since in case of fault injection part name may not be all_0_0_0 but all_1_1_0
clickhouse-disks -C "$config" --disk s3_plain_disk list --recursive "${path:?}" | tail -n+2 | sed 's/all_[^_]*_[^_]*_0/all_X_X_X/g'
clickhouse-disks -C "$config" --disk s3_plain_disk --query "list --recursive $path" | tail -n+2 | sed 's/all_[^_]*_[^_]*_0/all_X_X_X/g'
$CLICKHOUSE_CLIENT -nm -q "
detach table data_read;
detach table data_write;
"
echo "Files after DETACH TABLE"
clickhouse-disks -C "$config" --disk s3_plain_disk list --recursive "$path" | tail -n+2 | sed 's/all_[^_]*_[^_]*_0/all_X_X_X/g'
clickhouse-disks -C "$config" --disk s3_plain_disk --query "list --recursive $path" | tail -n+2 | sed 's/all_[^_]*_[^_]*_0/all_X_X_X/g'
# metadata file is left
$CLICKHOUSE_CLIENT --force_remove_data_recursively_on_drop=1 -q "drop database if exists $CLICKHOUSE_DATABASE"