#include <Client.h>

#include <base/defines.h>

#include <Client/ConnectionString.h>

#include <Core/Protocol.h>

#include <Core/Settings.h>

/// musl defines stderr as (stderr) which is a self-referential macro

#pragma clang diagnostic ignored "-Wdisabled-macro-expansion"

#include <boost/algorithm/string/replace.hpp>

#include <boost/program_options.hpp>

#include <Common/Config/parseConnectionCredentials.h>

#include <Common/ThreadStatus.h>

#include <Access/AccessControl.h>

#include <Columns/ColumnString.h>

#include <Common/Config/ConfigProcessor.h>

#include <Common/Config/getClientConfigPath.h>

#include <Common/CurrentThread.h>

#include <Common/DateLUT.h>

#include <Common/DateLUTImpl.h>

#include <Common/QueryScope.h>

#include <Common/Exception.h>

#include <Common/TerminalSize.h>

#include <Common/config_version.h>

#include <Common/formatReadable.h>

#include <IO/ReadBufferFromString.h>

#include <IO/ReadHelpers.h>

#include <IO/WriteBufferFromOStream.h>

#include <IO/WriteHelpers.h>

#include <Interpreters/Context.h>

#include <Client/JWTProvider.h>

#include <Client/ClientBaseHelpers.h>

#include <AggregateFunctions/registerAggregateFunctions.h>

#include <Formats/FormatFactory.h>

#include <Formats/registerFormats.h>

#include <Functions/registerFunctions.h>

#include <Storages/MergeTree/MergeTreeSettings.h>

#include <Poco/Util/Application.h>

#include <Poco/URI.h>

#include <filesystem>

#include "config.h"

#if USE_BUZZHOUSE

# include <Client/BuzzHouse/Generator/ExternalIntegrations.h>

# include <Client/BuzzHouse/Generator/FuzzConfig.h>

#endif

namespace fs = std::filesystem;

using namespace std::literals;

namespace DB

{

namespace Setting

{

extern const SettingsDialect dialect;

extern const SettingsBool use_client_time_zone;

}

namespace ErrorCodes

{

extern const int BAD_ARGUMENTS;

extern const int UNKNOWN_PACKET_FROM_SERVER;

extern const int NETWORK_ERROR;

extern const int AUTHENTICATION_FAILED;

extern const int REQUIRED_SECOND_FACTOR;

extern const int REQUIRED_PASSWORD;

extern const int USER_EXPIRED;

}

Client::Client()

{

fuzzer = QueryFuzzer(randomSeed(), &std::cout, &std::cerr);

}

Client::~Client() = default;

void Client::processError(std::string_view query) const

{

if (server_exception)

{

fmt::print(

stderr,

"Received exception from server (version {}):\n{}\n",

server_version,

getExceptionMessageForLogging(*server_exception, print_stack_trace, true));

if (server_exception->code() == ErrorCodes::USER_EXPIRED)

{

server_exception->rethrow();

}

if (is_interactive)

{

fmt::print(stderr, "\n");

}

else

{

fmt::print(stderr, "(query: {})\n", query);

}

}

if (client_exception)

{

fmt::print(stderr, "Error on processing query: {}\n", client_exception->message());

if (is_interactive)

{

fmt::print(stderr, "\n");

}

else

{

fmt::print(stderr, "(query: {})\n", query);

}

}

// A debug check -- at least some exception must be set, if the error

// flag is set, and vice versa.

chassert(have_error == (client_exception || server_exception));

}

void Client::showWarnings()

{

try

{

std::vector<String> messages = loadWarningMessages();

if (!messages.empty())

{

output_stream << "Warnings:" << std::endl;

for (const auto & message : messages)

output_stream << " * " << message << std::endl;

output_stream << std::endl;

}

}

catch (const std::exception &) // NOLINT(bugprone-empty-catch)

{

}

}

/// Make query to get all server warnings

std::vector<String> Client::loadWarningMessages()

{

/// Older server versions cannot execute the query loading warnings.

constexpr UInt64 min_server_revision_to_load_warnings = DBMS_MIN_PROTOCOL_VERSION_WITH_VIEW_IF_PERMITTED;

if (server_revision < min_server_revision_to_load_warnings)

return {};

std::vector<String> messages;

connection->sendQuery(connection_parameters.timeouts,

"SELECT * FROM viewIfPermitted(SELECT message FROM system.warnings ELSE null('message String'))",

{} /* query_parameters */,

"" /* query_id */,

QueryProcessingStage::Complete,

&client_context->getSettingsRef(),

&client_context->getClientInfo(), false, {}, {});

while (true)

{

Packet packet = connection->receivePacket();

switch (packet.type)

{

case Protocol::Server::Data:

if (!packet.block.empty())

{

const ColumnString & column = typeid_cast<const ColumnString &>(*packet.block.getByPosition(0).column);

size_t rows = packet.block.rows();

for (size_t i = 0; i < rows; ++i)

messages.emplace_back(column[i].safeGet<String>());

}

continue;

case Protocol::Server::Progress:

case Protocol::Server::ProfileInfo:

case Protocol::Server::Totals:

case Protocol::Server::Extremes:

case Protocol::Server::Log:

continue;

case Protocol::Server::Exception:

packet.exception->rethrow();

return messages;

case Protocol::Server::EndOfStream:

return messages;

case Protocol::Server::ProfileEvents:

continue;

default:

throw Exception(

ErrorCodes::UNKNOWN_PACKET_FROM_SERVER, "Unknown packet {} from server {}", packet.type, connection->getDescription());

}

}

}

Poco::Util::LayeredConfiguration & Client::getClientConfiguration()

{

return config();

}

void Client::initialize(Poco::Util::Application & self)

{

Poco::Util::Application::initialize(self);

const char * home_path_cstr = getenv("HOME"); // NOLINT(concurrency-mt-unsafe)

if (home_path_cstr)

home_path = home_path_cstr;

const char * env_host = getenv("CLICKHOUSE_HOST"); // NOLINT(concurrency-mt-unsafe)

std::optional<std::string> config_path;

if (config().has("config-file"))

config_path.emplace(config().getString("config-file"));

else

config_path = getClientConfigPath(home_path);

if (config_path.has_value())

{

ConfigProcessor config_processor(*config_path);

auto loaded_config = config_processor.loadConfig();

auto & configuration = *loaded_config.configuration;

std::string default_host;

if (!hosts_and_ports.empty())

default_host = hosts_and_ports.front().host;

else if (config().has("host"))

default_host = config().getString("host");

else if (configuration.has("host"))

default_host = configuration.getString("host");

else if (env_host)

default_host = env_host;

else

default_host = "localhost";

std::optional<std::string> connection_name;

if (config().has("connection"))

connection_name.emplace(config().getString("connection"));

/// Connection credentials overrides should be set via loaded_config.configuration to have proper order.

auto overrides = parseConnectionsCredentials(configuration, default_host, connection_name);

if (overrides.hostname.has_value())

configuration.setString("host", overrides.hostname.value());

if (overrides.port.has_value())

configuration.setInt("port", overrides.port.value());

if (overrides.secure.has_value())

{

if (overrides.secure.value())

configuration.setBool("secure", true);

else

configuration.setBool("no-secure", true);

}

if (overrides.user.has_value())

configuration.setString("user", overrides.user.value());

if (overrides.password.has_value())

configuration.setString("password", overrides.password.value());

if (overrides.database.has_value())

configuration.setString("database", overrides.database.value());

if (overrides.history_file.has_value())

{

auto history_file = overrides.history_file.value();

if (history_file.starts_with("~/") && !home_path.empty())

history_file = home_path / history_file.substr(2);

configuration.setString("history_file", history_file);

}

if (overrides.history_max_entries.has_value())

configuration.setUInt("history_max_entries", overrides.history_max_entries.value());

if (overrides.accept_invalid_certificate.has_value())

configuration.setBool("accept-invalid-certificate", overrides.accept_invalid_certificate.value());

if (overrides.prompt.has_value())

configuration.setString("prompt", overrides.prompt.value());

config().add(loaded_config.configuration);

#if USE_JWT_CPP && USE_SSL

/// If config file has user/password credentials, don't use auto-detected OAuth login for cloud endpoints

if (login_was_auto_added &&

(loaded_config.configuration->has("user") || loaded_config.configuration->has("password")))

{

/// Config file has auth credentials, so disable the auto-added login flag

config().setBool("login", false);

}

#endif

}

else if (config().has("connection"))

throw Exception(ErrorCodes::BAD_ARGUMENTS, "--connection was specified, but config does not exist");

if (config().has("accept-invalid-certificate"))

{

config().setString("openSSL.client.invalidCertificateHandler.name", "AcceptCertificateHandler");

config().setString("openSSL.client.verificationMode", "none");

}

/** getenv is thread-safe in Linux glibc and in all sane libc implementations.

* But the standard does not guarantee that subsequent calls will not rewrite the value by returned pointer.

*

* man getenv:

*

* As typically implemented, getenv() returns a pointer to a string within the environment list.

* The caller must take care not to modify this string, since that would change the environment of

* the process.

*

* The implementation of getenv() is not required to be reentrant. The string pointed to by the return value of getenv()

* may be statically allocated, and can be modified by a subsequent call to getenv(), putenv(3), setenv(3), or unsetenv(3).

*/

const char * env_user = getenv("CLICKHOUSE_USER"); // NOLINT(concurrency-mt-unsafe)

if (env_user && !config().has("user"))

config().setString("user", env_user);

const char * env_password = getenv("CLICKHOUSE_PASSWORD"); // NOLINT(concurrency-mt-unsafe)

if (env_password && !config().has("password"))

config().setString("password", env_password);

if (env_host && !config().has("host"))

config().setString("host", env_host);

/// settings and limits could be specified in config file, but passed settings has higher priority

for (const auto & setting : client_context->getSettingsRef().getUnchangedNames())

{

String name{setting};

if (config().has(name))

client_context->setSetting(name, config().getString(name));

}

/// Set path for format schema files

if (config().has("format_schema_path"))

client_context->setFormatSchemaPath(fs::weakly_canonical(config().getString("format_schema_path")));

/// Set the path for google proto files

if (config().has("google_protos_path"))

client_context->setGoogleProtosPath(fs::weakly_canonical(config().getString("google_protos_path")));

/// Use <server_client_version_message/> unless --server-client-version-message is specified

if (!config().has("no-server-client-version-message") && !config().getBool("server_client_version_message", true))

config().setBool("no-server-client-version-message", true);

/// Use <warnings/> unless --no-warnings is specified

if (!config().has("no-warnings") && !config().getBool("warnings", true))

config().setBool("no-warnings", true);

}

int Client::main(const std::vector<std::string> & /*args*/)

try

{

setupSignalHandler();

output_stream << std::fixed << std::setprecision(3);

error_stream << std::fixed << std::setprecision(3);

registerFormats();

registerFunctions();

registerAggregateFunctions();

processConfig();

adjustSettings(client_context);

initTTYBuffer(

toProgressOption(config().getString("progress", "default")), toProgressOption(config().getString("progress-table", "default")));

initKeystrokeInterceptor();

/// Includes delayed_interactive.

if (is_interactive)

{

clearTerminal();

showClientVersion();

}

#if USE_JWT_CPP && USE_SSL

if (config().getBool("login", false))

{

login();

}

#endif

bool asked_password = false;

bool asked_2fa = false;

for (;;)

{

try

{

connect();

break;

}

catch (const Exception & e)

{

auto code = e.code();

bool should_ask_password = !asked_password && is_interactive &&

(code == ErrorCodes::AUTHENTICATION_FAILED || code == ErrorCodes::REQUIRED_PASSWORD) &&

!config().has("password") && !config().getBool("ask-password", false) &&

!config().has("ssh-key-file");

if (should_ask_password)

{

asked_password = true;

config().setBool("ask-password", true);

continue;

}

bool should_ask_2fa = !asked_2fa && (code == ErrorCodes::REQUIRED_SECOND_FACTOR) &&

(config().getBool("ask-password", false) || is_interactive) && !config().has("one-time-password");

if (should_ask_2fa)

{

asked_2fa = true;

if (!connection_parameters.password.empty())

config().setString("password", connection_parameters.password);

config().setBool("ask-password", false);

config().setBool("ask-password-2fa", true);

continue;

}

throw;

}

}

/// Show warnings at the beginning of connection.

if (is_interactive && !config().has("no-warnings"))

showWarnings();

/// Set user password complexity rules

auto & access_control = client_context->getAccessControl();

access_control.setPasswordComplexityRules(connection->getPasswordComplexityRules());

if (is_interactive && !delayed_interactive && !buzz_house)

{

runInteractive();

}

else

{

connection->setDefaultDatabase(connection_parameters.default_database);

runNonInteractive();

// If exception code isn't zero, we should return non-zero return

// code anyway.

const auto * exception = server_exception ? server_exception.get() : client_exception.get();

if (exception)

{

return static_cast<UInt8>(exception->code()) ? exception->code() : -1;

}

if (have_error)

{

// Shouldn't be set without an exception, but check it just in

// case so that at least we don't lose an error.

return -1;

}

if (delayed_interactive)

runInteractive();

}

return 0;

}

catch (Exception & e)

{

bool need_print_stack_trace = config().getBool("stacktrace", false) && e.code() != ErrorCodes::NETWORK_ERROR;

std::cerr << getExceptionMessageForLogging(e, need_print_stack_trace, true) << std::endl << std::endl;

/// If exception code isn't zero, we should return non-zero return code anyway.

return static_cast<UInt8>(e.code()) ? e.code() : -1;

}

catch (...)

{

std::cerr << getCurrentExceptionMessage(false) << std::endl;

return getCurrentExceptionCode();

}

#if USE_JWT_CPP && USE_SSL

void Client::login()

{

std::string host = hosts_and_ports.front().host;

std::string auth_url = getClientConfiguration().getString("oauth-url", "");

std::string client_id = getClientConfiguration().getString("oauth-client-id", "");

std::string audience = getClientConfiguration().getString("oauth-audience", "");

if ((auth_url.empty() || client_id.empty()) && !isCloudEndpoint(host))

{

throw Exception(

ErrorCodes::BAD_ARGUMENTS,

"Could not retrieve authentication endpoints for host '{}'. Please specify --oauth-url and --oauth-client-id if you are "

"not using ClickHouse Cloud.",

host);

}

jwt_provider = createJwtProvider(auth_url, client_id, audience, host, output_stream, error_stream);

if (jwt_provider)

{

std::string jwt = jwt_provider->getJWT();

if (!jwt.empty())

{

getClientConfiguration().setString("jwt", jwt);

}

else

{

throw Exception(ErrorCodes::AUTHENTICATION_FAILED, "Login failed. Please check your credentials and try again.");

}

}

}

#endif

void Client::connect()

{

String server_name;

UInt64 server_version_major = 0;

UInt64 server_version_minor = 0;

UInt64 server_version_patch = 0;

/// Capture the client local time zone before the branch below may switch the process default

/// to the server time zone. `serverTimezoneInstance()` reads the process default directly and

/// ignores `session_timezone`; `instance()` would fold in an explicit `--session_timezone` and

/// cache the wrong zone. `connect()` can run again on reconnect, so only capture once.

if (client_local_timezone.empty())

client_local_timezone = DateLUT::serverTimezoneInstance().getTimeZone();

if (hosts_and_ports.empty())

{

String host = config().getString("host", "localhost");

UInt16 port = ConnectionParameters::getPortFromConfig(config(), host);

hosts_and_ports.emplace_back(HostAndPort{host, port});

}

for (size_t attempted_address_index = 0; attempted_address_index < hosts_and_ports.size(); ++attempted_address_index)

{

try

{

const auto host = ConnectionParameters::Host{hosts_and_ports[attempted_address_index].host};

const auto database = ConnectionParameters::Database{default_database};

connection_parameters = ConnectionParameters(

config(), host, database, hosts_and_ports[attempted_address_index].port);

#if USE_JWT_CPP && USE_SSL

connection_parameters.jwt_provider = jwt_provider;

#endif

if (is_interactive)

output_stream << "Connecting to "

<< (!connection_parameters.default_database.empty()

? "database " + connection_parameters.default_database + " at "

: "")

<< connection_parameters.host << ":" << connection_parameters.port

<< (!connection_parameters.user.empty() ? " as user " + connection_parameters.user : "") << "." << std::endl;

connection = Connection::createConnection(connection_parameters, client_context);

if (max_client_network_bandwidth)

{

ThrottlerPtr throttler = std::make_shared<Throttler>(max_client_network_bandwidth, 0, "");

connection->setThrottler(throttler);

}

connection->getServerVersion(

connection_parameters.timeouts,

server_name,

server_version_major,

server_version_minor,

server_version_patch,

server_revision);

config().setString("host", connection_parameters.host);

config().setInt("port", connection_parameters.port);

settings_from_server = assert_cast<Connection &>(*connection).settingsFromServer();

break;

}

catch (Exception & e)

{

/// This problem can't be fixed with reconnection so it is not attempted

if (e.code() == ErrorCodes::AUTHENTICATION_FAILED || e.code() == ErrorCodes::REQUIRED_PASSWORD)

throw;

if (attempted_address_index == hosts_and_ports.size() - 1)

throw;

if (is_interactive)

{

std::cerr << "Connection attempt to database at " << connection_parameters.host << ":" << connection_parameters.port

<< " resulted in failure" << std::endl

<< getExceptionMessageForLogging(e, false) << std::endl

<< "Attempting connection to the next provided address" << std::endl;

}

}

}

server_version = toString(server_version_major) + "." + toString(server_version_minor) + "." + toString(server_version_patch);

load_suggestions

= is_interactive && (server_revision >= Suggest::MIN_SERVER_REVISION) && !config().getBool("disable_suggestion", false);

wait_for_suggestions_to_load = config().getBool("wait_for_suggestions_to_load", false);

if (load_suggestions)

{

suggestion_limit = config().getInt("suggestion_limit");

}

server_display_name = connection->getServerDisplayName(connection_parameters.timeouts);

if (server_display_name.empty())

server_display_name = config().getString("host", "localhost");

if (is_interactive)

{

output_stream << "Connected to " << server_name << " server version " << server_version << "." << std::endl << std::endl;

#if not CLICKHOUSE_CLOUD

if (!config().has("no-server-client-version-message"))

{

auto client_version_tuple = std::make_tuple(VERSION_MAJOR, VERSION_MINOR, VERSION_PATCH);

auto server_version_tuple = std::make_tuple(server_version_major, server_version_minor, server_version_patch);

if (client_version_tuple < server_version_tuple)

{

output_stream << "ClickHouse client version is older than ClickHouse server. "

<< "It may lack support for new features." << std::endl

<< std::endl;

}

else if (client_version_tuple > server_version_tuple && server_display_name != "clickhouse-cloud")

{

output_stream << "ClickHouse server version is older than ClickHouse client. "

<< "It may indicate that the server is out of date and can be upgraded." << std::endl

<< std::endl;

}

}

#endif

}

if (!client_context->getSettingsRef()[Setting::use_client_time_zone])

{

const auto & time_zone = connection->getServerTimezone(connection_parameters.timeouts);

if (!time_zone.empty())

{

try

{

DateLUT::setDefaultTimezone(time_zone);

}

catch (...)

{

std::cerr << "Warning: could not switch to server time zone: " << time_zone

<< ", reason: " << getCurrentExceptionMessage(/* with_stacktrace = */ false) << std::endl

<< "Proceeding with local time zone." << std::endl

<< std::endl;

}

}

else

{

std::cerr << "Warning: could not determine server time zone. "

<< "Proceeding with local time zone." << std::endl

<< std::endl;

}

}

/// A custom prompt can be specified

/// - directly (possible as CLI parameter or in client.xml as top-level <prompt>...</prompt> or within client.xml's connection credentials)

/// - via prompt_by_server_display_name (only possible in client.xml as top-level <prompt>...</prompt>).

if (config().has("prompt"))

prompt = config().getString("prompt");

else if (config().has("prompt_by_server_display_name"))

{

if (config().has("prompt_by_server_display_name.default"))

prompt = config().getRawString("prompt_by_server_display_name.default");

Strings keys;

config().keys("prompt_by_server_display_name", keys);

for (const auto & key : keys)

{

if (key != "default" && server_display_name.contains(key))

{

prompt = config().getRawString("prompt_by_server_display_name." + key);

break;

}

}

}

else

{

prompt = "{display_name}";

}

/// Prompt may contain escape sequences including \e[ or \x1b[ sequences to set terminal color.

{

String prompt_escaped;

ReadBufferFromString in(prompt);

readEscapedString(prompt_escaped, in);

prompt = prompt_escaped;

}

/// Substitute placeholders in the form of {name}:

const std::map<String, String> prompt_substitutions{

{"host", connection_parameters.host},

{"port", toString(connection_parameters.port)},

{"user", connection_parameters.user},

{"display_name", server_display_name},

};

for (const auto & [key, value] : prompt_substitutions)

boost::replace_all(prompt, "{" + key + "}", value);

prompt = appendSmileyIfNeeded(prompt);

}

// Prints changed settings to stderr. Useful for debugging fuzzing failures.

void Client::printChangedSettings() const

{

auto print_changes = [](const auto & changes, std::string_view settings_name)

{

if (!changes.empty())

{

fmt::print(stderr, "Changed {}: ", settings_name);

for (size_t i = 0; i < changes.size(); ++i)

{

if (i)

fmt::print(stderr, ", ");

fmt::print(stderr, "{} = '{}'", changes[i].name, fieldToString(changes[i].value));

}

fmt::print(stderr, "\n");

}

else

{

fmt::print(stderr, "No changed {}.\n", settings_name);

}

};

print_changes(client_context->getSettingsRef().changes(), "settings");

print_changes(cmd_merge_tree_settings->changes(), "MergeTree settings");

}

String Client::getHelpHeader() const

{

return fmt::format(

"Usage: {0} [--query <query>]\n"

"{0} is a client application that is used to connect to ClickHouse.\n\n"

"It can run queries as a command line tool if you pass queries as an argument\n"

"or as an interactive client.\n"

"Queries can run one at a time, or in a multiquery mode.\n"

"To change settings you may use SET statements and SETTINGS clause\n"

"in queries or set them for a session with corresponding arguments.\n"

"'{0}' command will try to connect to clickhouse-server running\n"

"on the same server. If you have credentials set up, pass them with\n"

"--user <username> --password <password> or with --ask-password argument\n"

"that will open command prompt.\n\n"

"Connect to tcp native port (9000) without encryption:\n"

" {0} --host clickhouse.example.com --password mysecretpassword\n"

"Connect to secure endpoint:\n"

" {0} --secure --host clickhouse.example.com --password mysecretpassword\n",

app_name);

}

String Client::getHelpFooter() const

{

return fmt::format(

"Note: if clickhouse is installed, you can use 'clickhouse-client' invocation with a dash.\n\n"

"Example printing current longest running query on a server:\n"

" {0} --query \\\n"

" 'SELECT * FROM system.processes ORDER BY elapsed LIMIT 1 FORMAT Vertical'\n"

"Example creating table and inserting data:\n"

" {0} --multiquery --query \\\n"

" 'CREATE TABLE t (a Int) ENGINE = Memory; INSERT INTO t VALUES (1), (2), (3)'\n",

app_name);

}

void Client::printHelpMessage(const OptionsDescription & options_description)

{

output_stream << getHelpHeader() << "\n";

if (options_description.main_description.has_value())

output_stream << options_description.main_description.value() << "\n";

if (options_description.external_description.has_value())

output_stream << options_description.external_description.value() << "\n";

if (options_description.hosts_and_ports_description.has_value())

output_stream << options_description.hosts_and_ports_description.value() << "\n";

output_stream << "All settings are documented at https://clickhouse.com/docs/operations/settings/settings.\n";

output_stream << getHelpFooter() << "\n";

output_stream << "In addition, --param_name=value can be specified for substitution of parameters for parameterized queries.\n";

output_stream << "\nSee also: https://clickhouse.com/docs/en/integrations/sql-clients/cli\n";

}

void Client::addExtraOptions(OptionsDescription & options_description)

{

/// Main commandline options related to client functionality and all parameters from Settings.

options_description.main_description->add_options()

("config,c", po::value<std::string>(), "config-file path (another shorthand)")

("connection", po::value<std::string>(), "connection to use (from the client config), by default connection name is hostname")

("secure,s", "Use TLS connection")

("tls-sni-override", po::value<std::string>(), "Override the SNI host name used for TLS connections")

("no-secure", "Don't use TLS connection")

("user,u", po::value<std::string>()->default_value("default"), "user")

("password", po::value<std::string>(), "password")

("ask-password", "ask-password")

("ssh-key-file", po::value<std::string>(), "File containing the SSH private key for authenticate with the server.")

("ssh-key-passphrase", po::value<std::string>(), "Passphrase for the SSH private key specified by --ssh-key-file.")

("quota_key", po::value<std::string>(), "A string to differentiate quotas when the user have keyed quotas configured on server")

("jwt", po::value<std::string>(), "Use JWT for authentication")

("one-time-password", po::value<std::string>(), "Time-based one-time password (TOTP) for two-factor authentication")

#if USE_JWT_CPP && USE_SSL

("login", po::bool_switch(), "Use OAuth 2.0 to login")

("oauth-url", po::value<std::string>(), "The base URL for the OAuth 2.0 authorization server")

("oauth-client-id", po::value<std::string>(), "The client ID for the OAuth 2.0 application")

("oauth-audience", po::value<std::string>(), "The audience for the OAuth 2.0 token")

#endif

("max_client_network_bandwidth",

po::value<int>(),

"the maximum speed of data exchange over the network for the client in bytes per second.")

("compression",

po::value<bool>(),

"enable or disable compression (enabled by default for remote communication and disabled for localhost communication).")

("query-fuzzer-runs",

po::value<int>()->default_value(0),

"After executing every SELECT query, do random mutations in it and run again specified number of times. This is used for "

"testing to discover unexpected corner cases.")

("create-query-fuzzer-runs", po::value<int>()->default_value(0), "")

("buzz-house-config", po::value<std::string>(), "Path to configuration file for BuzzHouse")

("interleave-queries-file",

po::value<std::vector<std::string>>()->multitoken(),

"file path with queries to execute before every file from 'queries-file'; multiple files can be specified (--queries-file "

"file1 file2...); this is needed to enable more aggressive fuzzing of newly added tests (see 'query-fuzzer-runs' option)")

("opentelemetry-traceparent",

po::value<std::string>(),

"OpenTelemetry traceparent header as described by W3C Trace Context recommendation")

("opentelemetry-tracestate",

po::value<std::string>(),

"OpenTelemetry tracestate header as described by W3C Trace Context recommendation")

("no-warnings", "disable warnings when client connects to server")

("no-server-client-version-message", "suppress server-client version mismatch message")

/// TODO: Left for compatibility as it's used in upgrade check, remove after next release and use server setting ignore_drop_queries_probability

("fake-drop", "Ignore all DROP queries, should be used only for testing")

("accept-invalid-certificate",

"Ignore certificate verification errors, equal to config parameters "

"openSSL.client.invalidCertificateHandler.name=AcceptCertificateHandler and openSSL.client.verificationMode=none")

("inline-insert-data",

"Send INSERT data as is in the query text instead of converting it to blocks in the native format. "

"The server will parse the inline data itself, avoiding the round-trip to receive table structure.");

/// Commandline options related to external tables.

options_description.external_description.emplace(createOptionsDescription("External tables options", terminal_width));

options_description.external_description->add_options()

("external", "marks the beginning of a clause. You may have multiple sections like this, for the number of tables being transmitted")

("file", po::value<std::string>(), "data file or - for stdin")

("name", po::value<std::string>()->default_value("_data"), "name of the table")

("format", po::value<std::string>()->default_value("TabSeparated"), "data format")

("structure", po::value<std::string>(), "structure")

("types", po::value<std::string>(), "types");

/// Commandline options related to hosts and ports.

options_description.hosts_and_ports_description.emplace(createOptionsDescription("Hosts and ports options", terminal_width));

options_description.hosts_and_ports_description->add_options()

("host,h", po::value<String>()->default_value("localhost"),

"Server hostname. Multiple hosts can be passed via multiple arguments"

"Example of usage: '--host host1 --host host2 --port port2 --host host3 ...'"

"Each '--port port' will be attached to the last seen host that doesn't have a port yet,"

"if there is no such host, the port will be attached to the next first host or to default host.")

("port", po::value<UInt16>(), "server ports");

}

void Client::processOptions(

const OptionsDescription & options_description,

const CommandLineOptions & options,

const std::vector<Arguments> & external_tables_arguments,

const std::vector<Arguments> & hosts_and_ports_arguments)

{

namespace po = boost::program_options;

size_t number_of_external_tables_with_stdin_source = 0;

for (size_t i = 0; i < external_tables_arguments.size(); ++i)

{

/// Parse commandline options related to external tables.

po::parsed_options parsed_tables

= po::command_line_parser(external_tables_arguments[i]).options(options_description.external_description.value()).run();

po::variables_map external_options;

po::store(parsed_tables, external_options);

try

{

external_tables.emplace_back(external_options);

if (external_tables.back().file == "-")

++number_of_external_tables_with_stdin_source;

if (number_of_external_tables_with_stdin_source > 1)

throw Exception(ErrorCodes::BAD_ARGUMENTS, "Two or more external tables has stdin (-) set as --file field");

}

catch (Exception & e)

{

std::cerr << getExceptionMessageForLogging(e, false) << std::endl;

std::cerr << "Table №" << i << std::endl << std::endl;

/// Avoid the case when error exit code can possibly overflow to normal (zero).

auto exit_code = e.code() % 256;

if (exit_code == 0)

exit_code = 255;

_exit(exit_code);

}

}

for (const auto & hosts_and_ports_argument : hosts_and_ports_arguments)

{

/// Parse commandline options related to external tables.

po::parsed_options parsed_hosts_and_ports

= po::command_line_parser(hosts_and_ports_argument).options(options_description.hosts_and_ports_description.value()).run();

po::variables_map host_and_port_options;

po::store(parsed_hosts_and_ports, host_and_port_options);

std::string host = host_and_port_options["host"].as<std::string>();

std::optional<UInt16> port

= !host_and_port_options["port"].empty() ? std::make_optional(host_and_port_options["port"].as<UInt16>()) : std::nullopt;

hosts_and_ports.emplace_back(HostAndPort{host, port});

}

send_external_tables = true;

shared_context = Context::createShared();

global_context = Context::createGlobal(shared_context.get());

global_context->makeGlobalContext();

global_context->setApplicationType(Context::ApplicationType::CLIENT);

global_context->setSettings(*cmd_settings);

/// Copy settings-related program options to config.

/// TODO: Is this code necessary?

global_context->getSettingsRef().addToClientOptions(config(), options, allow_repeated_settings);

if (options.contains("config-file") && options.contains("config"))

throw Exception(ErrorCodes::BAD_ARGUMENTS, "Two or more configuration files referenced in arguments");

if (options.contains("config"))

config().setString("config-file", options["config"].as<std::string>());

if (options.contains("connection"))

config().setString("connection", options["connection"].as<std::string>());

if (options.contains("interleave-queries-file"))

interleave_queries_files = options["interleave-queries-file"].as<std::vector<std::string>>();

if (options.contains("secure"))

config().setBool("secure", true);

if (options.contains("tls-sni-override"))

config().setString("tls-sni-override", options["tls-sni-override"].as<std::string>());

if (options.contains("no-secure"))

config().setBool("no-secure", true);

if (options.contains("user") && !options["user"].defaulted())

config().setString("user", options["user"].as<std::string>());

if (options.contains("password"))

config().setString("password", options["password"].as<std::string>());

if (options.contains("ask-password"))

config().setBool("ask-password", true);

if (options.contains("one-time-password"))

config().setString("one-time-password", options["one-time-password"].as<std::string>());

if (options.contains("ssh-key-file"))

config().setString("ssh-key-file", options["ssh-key-file"].as<std::string>());

if (options.contains("ssh-key-passphrase"))

config().setString("ssh-key-passphrase", options["ssh-key-passphrase"].as<std::string>());

if (options.contains("quota_key"))

config().setString("quota_key", options["quota_key"].as<std::string>());

if (options.contains("max_client_network_bandwidth"))

max_client_network_bandwidth = options["max_client_network_bandwidth"].as<int>();

if (options.contains("compression"))

config().setBool("compression", options["compression"].as<bool>());

if (options.contains("no-warnings"))

config().setBool("no-warnings", true);

if (options.contains("no-server-client-version-message"))

config().setBool("no-server-client-version-message", true);

if (options.contains("fake-drop"))

config().setString("ignore_drop_queries_probability", "1");

if (options.contains("inline-insert-data"))

config().setBool("inline-insert-data", true);

if (options.contains("jwt"))

{

if (!options["user"].defaulted())

throw Exception(ErrorCodes::BAD_ARGUMENTS, "User and JWT flags can't be specified together");

config().setString("jwt", options["jwt"].as<std::string>());

config().setString("user", "");

}

#if USE_JWT_CPP && USE_SSL

if (options["login"].as<bool>())

{

if (!options["user"].defaulted())

throw Exception(ErrorCodes::BAD_ARGUMENTS, "User and login flags can't be specified together");

if (config().has("jwt"))

throw Exception(ErrorCodes::BAD_ARGUMENTS, "JWT and login flags can't be specified together");

config().setBool("login", true);

config().setString("user", "");

}

if (options.contains("oauth-url"))

config().setString("oauth-url", options["oauth-url"].as<std::string>());

if (options.contains("oauth-client-id"))

config().setString("oauth-client-id", options["oauth-client-id"].as<std::string>());

if (options.contains("oauth-audience"))

config().setString("oauth-audience", options["oauth-audience"].as<std::string>());

#endif

if (options.contains("accept-invalid-certificate"))

{

config().setString("openSSL.client.invalidCertificateHandler.name", "AcceptCertificateHandler");