thevar1able/ClickHouse
1
0
Fork 0
mirror of https://github.com/ClickHouse/ClickHouse.git synced 2024-12-15 19:02:04 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'master' into remove-wrong-code

Browse Source
This commit is contained in:
Alexey Milovidov 2023-07-06 02:51:34 +02:00
commit 731dc1eba4
86 changed files with 2244 additions and 194 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
docker/test/integration/runner/Dockerfile
View File

@ -47,11 +47,13 @@ ENV TZ=Etc/UTC
RUN ln -snf /usr/share/zoneinfo/$TZ /etc/localtime && echo $TZ > /etc/timezone
ENV DOCKER_CHANNEL stable
# Unpin the docker version after the release 24.0.3 is released
# https://github.com/moby/moby/issues/45770#issuecomment-1618255130
RUN curl -fsSL https://download.docker.com/linux/ubuntu/gpg | apt-key add - \
&& add-apt-repository "deb https://download.docker.com/linux/ubuntu $(lsb_release -c -s) ${DOCKER_CHANNEL}" \
&& apt-get update \
&& env DEBIAN_FRONTEND=noninteractive apt-get install --yes \
docker-ce \
docker-ce='5:23.*' \
&& rm -rf \
/var/lib/apt/lists/* \
/var/cache/debconf \

3
docker/test/upgrade/run.sh
View File

@ -76,7 +76,8 @@ sudo mv /etc/clickhouse-server/config.d/keeper_port.xml.tmp /etc/clickhouse-serv
# But we still need default disk because some tables loaded only into it
sudo cat /etc/clickhouse-server/config.d/s3_storage_policy_by_default.xml \
| sed "s|<main><disk>s3</disk></main>|<main><disk>s3</disk></main><default><disk>default</disk></default>|" \
> /etc/clickhouse-server/config.d/s3_storage_policy_by_default.xml.tmp mv /etc/clickhouse-server/config.d/s3_storage_policy_by_default.xml.tmp /etc/clickhouse-server/config.d/s3_storage_policy_by_default.xml
> /etc/clickhouse-server/config.d/s3_storage_policy_by_default.xml.tmp
mv /etc/clickhouse-server/config.d/s3_storage_policy_by_default.xml.tmp /etc/clickhouse-server/config.d/s3_storage_policy_by_default.xml
sudo chown clickhouse /etc/clickhouse-server/config.d/s3_storage_policy_by_default.xml
sudo chgrp clickhouse /etc/clickhouse-server/config.d/s3_storage_policy_by_default.xml

11
docs/en/engines/table-engines/integrations/odbc.md
View File

@ -54,7 +54,7 @@ $ sudo mysql
``` sql
mysql> CREATE USER 'clickhouse'@'localhost' IDENTIFIED BY 'clickhouse';
mysql> GRANT ALL PRIVILEGES ON *.* TO 'clickhouse'@'clickhouse' WITH GRANT OPTION;
mysql> GRANT ALL PRIVILEGES ON *.* TO 'clickhouse'@'localhost' WITH GRANT OPTION;
```
Then configure the connection in `/etc/odbc.ini`.
@ -66,7 +66,7 @@ DRIVER = /usr/local/lib/libmyodbc5w.so
SERVER = 127.0.0.1
PORT = 3306
DATABASE = test
USERNAME = clickhouse
USER = clickhouse
PASSWORD = clickhouse
```
@ -83,6 +83,9 @@ $ isql -v mysqlconn
Table in MySQL:
``` text
mysql> CREATE DATABASE test;
Query OK, 1 row affected (0,01 sec)
mysql> CREATE TABLE `test`.`test` (
-> `int_id` INT NOT NULL AUTO_INCREMENT,
-> `int_nullable` INT NULL DEFAULT NULL,
@ -91,10 +94,10 @@ mysql> CREATE TABLE `test`.`test` (
-> PRIMARY KEY (`int_id`));
Query OK, 0 rows affected (0,09 sec)
mysql> insert into test (`int_id`, `float`) VALUES (1,2);
mysql> insert into test.test (`int_id`, `float`) VALUES (1,2);
Query OK, 1 row affected (0,00 sec)
mysql> select * from test;
mysql> select * from test.test;
+------+----------+-----+----------+
| int_id | int_nullable | float | float_nullable |
+------+----------+-----+----------+

1
docs/en/operations/system-tables/index.md
View File

@ -13,6 +13,7 @@ System tables provide information about:
- Server states, processes, and environment.
- Server’s internal processes.
- Options used when the ClickHouse binary was built.
System tables:

2
docs/zh/sql-reference/functions/functions-for-nulls.md
View File

@ -192,7 +192,7 @@ SELECT coalesce(mail, phone, CAST(icq,'Nullable(String)')) FROM aBook
**返回值**
- 如果`x`不为`NULL`，返回非`Nullable`类型的原始值。
- 如果`x`为`NULL`，返回对应非`Nullable`类型的默认值。
- 如果`x`为`NULL`，则返回任意值。
**示例**

13
programs/local/LocalServer.cpp
View File

@ -8,7 +8,9 @@
#include <Poco/Logger.h>
#include <Poco/NullChannel.h>
#include <Poco/SimpleFileChannel.h>
#include <Databases/DatabaseFilesystem.h>
#include <Databases/DatabaseMemory.h>
#include <Databases/DatabasesOverlay.h>
#include <Storages/System/attachSystemTables.h>
#include <Storages/System/attachInformationSchemaTables.h>
#include <Interpreters/DatabaseCatalog.h>
@ -50,6 +52,8 @@
#include <base/argsToConfig.h>
#include <filesystem>
#include "config.h"
#if defined(FUZZING_MODE)
#include <Functions/getFuzzerData.h>
#endif
@ -170,6 +174,13 @@ static DatabasePtr createMemoryDatabaseIfNotExists(ContextPtr context, const Str
return system_database;
}
static DatabasePtr createClickHouseLocalDatabaseOverlay(const String & name_, ContextPtr context_)
{
auto databaseCombiner = std::make_shared<DatabasesOverlay>(name_, context_);
databaseCombiner->registerNextDatabase(std::make_shared<DatabaseFilesystem>(name_, "", context_));
databaseCombiner->registerNextDatabase(std::make_shared<DatabaseMemory>(name_, context_));
return databaseCombiner;
}
/// If path is specified and not empty, will try to setup server environment and load existing metadata
void LocalServer::tryInitPath()
@ -669,7 +680,7 @@ void LocalServer::processConfig()
* if such tables will not be dropped, clickhouse-server will not be able to load them due to security reasons.
*/
std::string default_database = config().getString("default_database", "_local");
DatabaseCatalog::instance().attachDatabase(default_database, std::make_shared<DatabaseMemory>(default_database, global_context));
DatabaseCatalog::instance().attachDatabase(default_database, createClickHouseLocalDatabaseOverlay(default_database, global_context));
global_context->setCurrentDatabase(default_database);
applyCmdOptions(global_context);

36
programs/server/dashboard.html
View File

@ -12,7 +12,8 @@
--chart-background: white;
--shadow-color: rgba(0, 0, 0, 0.25);
--input-shadow-color: rgba(0, 255, 0, 1);
--error-color: white;
--error-color: red;
--auth-error-color: white;
--legend-background: rgba(255, 255, 255, 0.75);
--title-color: #666;
--text-color: black;
@ -258,7 +259,7 @@
width: 60%;
padding: .5rem;
color: var(--error-color);
color: var(--auth-error-color);
display: flex;
flex-flow: row nowrap;
@ -906,9 +907,9 @@ async function draw(idx, chart, url_params, query) {
if (error) {
const errorMatch = errorMessages.find(({ regex }) => error.match(regex))
if (errorMatch) {
const match = error.match(errorMatch.regex)
const message = errorMatch.messageFunc(match)
const match = error.match(errorMatch.regex)
const message = errorMatch.messageFunc(match)
if (message) {
const authError = new Error(message)
throw authError
}
@ -930,7 +931,7 @@ async function draw(idx, chart, url_params, query) {
let title_div = chart.querySelector('.title');
if (error) {
error_div.firstChild.data = error;
title_div.style.display = 'none';
title_div.style.display = 'none';
error_div.style.display = 'block';
return false;
} else {
@ -1019,13 +1020,15 @@ async function drawAll() {
firstLoad = false;
} else {
enableReloadButton();
enableRunButton();
}
if (!results.includes(false)) {
if (results.includes(true)) {
const element = document.querySelector('.inputs');
element.classList.remove('unconnected');
const add = document.querySelector('#add');
add.style.display = 'block';
} else {
}
else {
const charts = document.querySelector('#charts')
charts.style.height = '0px';
}
@ -1050,6 +1053,13 @@ function disableReloadButton() {
reloadButton.classList.add('disabled')
}
function disableRunButton() {
const runButton = document.getElementById('run')
runButton.value = 'Reloading...'
runButton.disabled = true
runButton.classList.add('disabled')
}
function enableReloadButton() {
const reloadButton = document.getElementById('reload')
reloadButton.value = 'Reload'
@ -1057,11 +1067,19 @@ function enableReloadButton() {
reloadButton.classList.remove('disabled')
}
function enableRunButton() {
const runButton = document.getElementById('run')
runButton.value = 'Ok'
runButton.disabled = false
runButton.classList.remove('disabled')
}
function reloadAll() {
updateParams();
drawAll();
saveState();
disableReloadButton()
disableReloadButton();
disableRunButton();
}
document.getElementById('params').onsubmit = function(event) {

5
src/Client/Suggest.cpp
View File

@ -101,9 +101,8 @@ static String getLoadSuggestionQuery(Int32 suggestion_limit, bool basic_suggesti
add_column("name", "columns", true, suggestion_limit);
}
/// FIXME: Forbid this query using new analyzer because of bug https://github.com/ClickHouse/ClickHouse/issues/50669
/// We should remove this restriction after resolving this bug.
query = "SELECT DISTINCT arrayJoin(extractAll(name, '[\\\\w_]{2,}')) AS res FROM (" + query + ") WHERE notEmpty(res) SETTINGS allow_experimental_analyzer=0";
/// FIXME: This query does not work with the new analyzer because of bug https://github.com/ClickHouse/ClickHouse/issues/50669
query = "SELECT DISTINCT arrayJoin(extractAll(name, '[\\\\w_]{2,}')) AS res FROM (" + query + ") WHERE notEmpty(res)";
return query;
}

10
src/Common/OptimizedRegularExpression.cpp
View File

@ -540,7 +540,7 @@ bool OptimizedRegularExpressionImpl<thread_safe>::match(const char * subject, si
}
}
return re2->Match(StringPieceType(subject, subject_size), 0, subject_size, RegexType::UNANCHORED, nullptr, 0);
return re2->Match({subject, subject_size}, 0, subject_size, RegexType::UNANCHORED, nullptr, 0);
}
}
@ -585,9 +585,9 @@ bool OptimizedRegularExpressionImpl<thread_safe>::match(const char * subject, si
return false;
}
StringPieceType piece;
std::string_view piece;
if (!RegexType::PartialMatch(StringPieceType(subject, subject_size), *re2, &piece))
if (!RegexType::PartialMatch({subject, subject_size}, *re2, &piece))
return false;
else
{
@ -652,10 +652,10 @@ unsigned OptimizedRegularExpressionImpl<thread_safe>::match(const char * subject
return 0;
}
DB::PODArrayWithStackMemory<StringPieceType, 128> pieces(limit);
DB::PODArrayWithStackMemory<std::string_view, 128> pieces(limit);
if (!re2->Match(
StringPieceType(subject, subject_size),
{subject, subject_size},
0,
subject_size,
RegexType::UNANCHORED,

1
src/Common/OptimizedRegularExpression.h
View File

@ -52,7 +52,6 @@ public:
using MatchVec = std::vector<Match>;
using RegexType = std::conditional_t<thread_safe, re2::RE2, re2_st::RE2>;
using StringPieceType = std::conditional_t<thread_safe, re2::StringPiece, re2_st::StringPiece>;
OptimizedRegularExpressionImpl(const std::string & regexp_, int options = 0); /// NOLINT
/// StringSearcher store pointers to required_substring, it must be updated on move.

3
src/Common/SensitiveDataMasker.cpp
View File

@ -5,7 +5,6 @@
#include <atomic>
#include <re2/re2.h>
#include <re2/stringpiece.h>
#include <Poco/Util/AbstractConfiguration.h>
@ -44,7 +43,7 @@ private:
const std::string regexp_string;
const RE2 regexp;
const re2::StringPiece replacement;
const std::string_view replacement;
#ifndef NDEBUG
mutable std::atomic<std::uint64_t> matches_count = 0;

9
src/Common/ThreadStatus.cpp
View File

@ -67,8 +67,8 @@ ThreadGroup::ThreadGroup()
: master_thread_id(CurrentThread::get().thread_id)
{}
ThreadStatus::ThreadStatus()
: thread_id{getThreadId()}
ThreadStatus::ThreadStatus(bool check_current_thread_on_destruction_)
: thread_id{getThreadId()}, check_current_thread_on_destruction(check_current_thread_on_destruction_)
{
last_rusage = std::make_unique<RUsageCounters>();
@ -201,8 +201,11 @@ ThreadStatus::~ThreadStatus()
/// Only change current_thread if it's currently being used by this ThreadStatus
/// For example, PushingToViews chain creates and deletes ThreadStatus instances while running in the main query thread
if (current_thread == this)
if (check_current_thread_on_destruction)
{
assert(current_thread == this);
current_thread = nullptr;
}
}
void ThreadStatus::updatePerformanceCounters()

4
src/Common/ThreadStatus.h
View File

@ -224,8 +224,10 @@ private:
Poco::Logger * log = nullptr;
bool check_current_thread_on_destruction;
public:
ThreadStatus();
explicit ThreadStatus(bool check_current_thread_on_destruction_ = true);
~ThreadStatus();
ThreadGroupPtr getThreadGroup() const;

7
src/Common/parseGlobs.cpp
View File

@ -3,7 +3,6 @@
#include <IO/ReadBufferFromString.h>
#include <IO/Operators.h>
#include <re2/re2.h>
#include <re2/stringpiece.h>
#include <algorithm>
#include <sstream>
#include <iomanip>
@ -33,14 +32,14 @@ std::string makeRegexpPatternFromGlobs(const std::string & initial_str_with_glob
std::string escaped_with_globs = buf_for_escaping.str();
static const re2::RE2 enum_or_range(R"({([\d]+\.\.[\d]+|[^{}*,]+,[^{}*]*[^{}*,])})"); /// regexp for {expr1,expr2,expr3} or {M..N}, where M and N - non-negative integers, expr's should be without "{", "}", "*" and ","
re2::StringPiece input(escaped_with_globs);
re2::StringPiece matched;
std::string_view input(escaped_with_globs);
std::string_view matched;
std::ostringstream oss_for_replacing; // STYLE_CHECK_ALLOW_STD_STRING_STREAM
oss_for_replacing.exceptions(std::ios::failbit);
size_t current_index = 0;
while (RE2::FindAndConsume(&input, enum_or_range, &matched))
{
std::string buffer{matched};
std::string buffer(matched);
oss_for_replacing << escaped_with_globs.substr(current_index, matched.data() - escaped_with_globs.data() - current_index - 1) << '(';
if (buffer.find(',') == std::string::npos)

1
src/Compression/CompressionCodecLZ4.cpp
View File

@ -42,7 +42,6 @@ private:
UInt32 getMaxCompressedDataSize(UInt32 uncompressed_size) const override;
mutable LZ4::PerformanceStatistics lz4_stat;
ASTPtr codec_desc;
};

9
src/Core/SettingsQuirks.cpp
View File

@ -1,10 +1,11 @@
#include <base/defines.h>
#include <Core/SettingsQuirks.h>
#include <Core/Settings.h>
#include <Poco/Environment.h>
#include <Poco/Platform.h>
#include <Common/VersionNumber.h>
#include <Common/logger_useful.h>
#include <cstdlib>
namespace
{
@ -71,6 +72,12 @@ void applySettingsQuirks(Settings & settings, Poco::Logger * log)
}
}
#if defined(THREAD_SANITIZER)
settings.use_hedged_requests.value = false;
if (log)
LOG_WARNING(log, "use_hedged_requests has been disabled for the build with Thread Sanitizer, because they are using fibers, leading to a failed assertion inside TSan");
#endif
if (!queryProfilerWorks())
{
if (settings.query_profiler_real_time_period_ns)

70
src/Databases/DatabaseFactory.cpp
View File

@ -3,6 +3,7 @@
#include <filesystem>
#include <Databases/DatabaseAtomic.h>
#include <Databases/DatabaseDictionary.h>
#include <Databases/DatabaseFilesystem.h>
#include <Databases/DatabaseLazy.h>
#include <Databases/DatabaseMemory.h>
#include <Databases/DatabaseOrdinary.h>
@ -47,6 +48,14 @@
#include <Databases/SQLite/DatabaseSQLite.h>
#endif
#if USE_AWS_S3
#include <Databases/DatabaseS3.h>
#endif
#if USE_HDFS
#include <Databases/DatabaseHDFS.h>
#endif
namespace fs = std::filesystem;
namespace DB
@ -131,13 +140,13 @@ DatabasePtr DatabaseFactory::getImpl(const ASTCreateQuery & create, const String
static const std::unordered_set<std::string_view> database_engines{"Ordinary", "Atomic", "Memory",
"Dictionary", "Lazy", "Replicated", "MySQL", "MaterializeMySQL", "MaterializedMySQL",
"PostgreSQL", "MaterializedPostgreSQL", "SQLite"};
"PostgreSQL", "MaterializedPostgreSQL", "SQLite", "Filesystem", "S3", "HDFS"};
if (!database_engines.contains(engine_name))
throw Exception(ErrorCodes::BAD_ARGUMENTS, "Database engine name `{}` does not exist", engine_name);
static const std::unordered_set<std::string_view> engines_with_arguments{"MySQL", "MaterializeMySQL", "MaterializedMySQL",
"Lazy", "Replicated", "PostgreSQL", "MaterializedPostgreSQL", "SQLite"};
"Lazy", "Replicated", "PostgreSQL", "MaterializedPostgreSQL", "SQLite", "Filesystem", "S3", "HDFS"};
static const std::unordered_set<std::string_view> engines_with_table_overrides{"MaterializeMySQL", "MaterializedMySQL", "MaterializedPostgreSQL"};
bool engine_may_have_arguments = engines_with_arguments.contains(engine_name);
@ -432,6 +441,63 @@ DatabasePtr DatabaseFactory::getImpl(const ASTCreateQuery & create, const String
}
#endif
else if (engine_name == "Filesystem")
{
const ASTFunction * engine = engine_define->engine;
/// If init_path is empty, then the current path will be used
std::string init_path;
if (engine->arguments && !engine->arguments->children.empty())
{
if (engine->arguments->children.size() != 1)
throw Exception(ErrorCodes::BAD_ARGUMENTS, "Filesystem database requires at most 1 argument: filesystem_path");
const auto & arguments = engine->arguments->children;
init_path = safeGetLiteralValue<String>(arguments[0], engine_name);
}
return std::make_shared<DatabaseFilesystem>(database_name, init_path, context);
}
#if USE_AWS_S3
else if (engine_name == "S3")
{
const ASTFunction * engine = engine_define->engine;
DatabaseS3::Configuration config;
if (engine->arguments && !engine->arguments->children.empty())
{
ASTs & engine_args = engine->arguments->children;
config = DatabaseS3::parseArguments(engine_args, context);
}
return std::make_shared<DatabaseS3>(database_name, config, context);
}
#endif
#if USE_HDFS
else if (engine_name == "HDFS")
{
const ASTFunction * engine = engine_define->engine;
/// If source_url is empty, then table name must contain full url
std::string source_url;
if (engine->arguments && !engine->arguments->children.empty())
{
if (engine->arguments->children.size() != 1)
throw Exception(ErrorCodes::BAD_ARGUMENTS, "HDFS database requires at most 1 argument: source_url");
const auto & arguments = engine->arguments->children;
source_url = safeGetLiteralValue<String>(arguments[0], engine_name);
}
return std::make_shared<DatabaseHDFS>(database_name, source_url, context);
}
#endif
throw Exception(ErrorCodes::UNKNOWN_DATABASE_ENGINE, "Unknown database engine: {}", engine_name);
}

245
src/Databases/DatabaseFilesystem.cpp Normal file
View File

@ -0,0 +1,245 @@
#include <Databases/DatabaseFilesystem.h>
#include <IO/Operators.h>
#include <IO/WriteBufferFromString.h>
#include <Interpreters/Context.h>
#include <Parsers/ASTCreateQuery.h>
#include <Parsers/ASTFunction.h>
#include <Parsers/ASTLiteral.h>
#include <Parsers/ParserCreateQuery.h>
#include <Parsers/parseQuery.h>
#include <Storages/IStorage.h>
#include <TableFunctions/TableFunctionFactory.h>
#include <Common/filesystemHelpers.h>
#include <filesystem>
namespace fs = std::filesystem;
namespace DB
{
namespace ErrorCodes
{
extern const int LOGICAL_ERROR;
extern const int UNKNOWN_TABLE;
extern const int PATH_ACCESS_DENIED;
extern const int BAD_ARGUMENTS;
extern const int FILE_DOESNT_EXIST;
}
DatabaseFilesystem::DatabaseFilesystem(const String & name_, const String & path_, ContextPtr context_)
: IDatabase(name_), WithContext(context_->getGlobalContext()), path(path_), log(&Poco::Logger::get("DatabaseFileSystem(" + name_ + ")"))
{
bool is_local = context_->getApplicationType() == Context::ApplicationType::LOCAL;
fs::path user_files_path = is_local ? "" : fs::canonical(getContext()->getUserFilesPath());
if (fs::path(path).is_relative())
{
path = user_files_path / path;
}
else if (!is_local && !pathStartsWith(fs::path(path), user_files_path))
{
throw Exception(ErrorCodes::BAD_ARGUMENTS,
"Path must be inside user-files path: {}", user_files_path.string());
}
path = fs::absolute(path).lexically_normal();
if (!fs::exists(path))
throw Exception(ErrorCodes::BAD_ARGUMENTS, "Path does not exist: {}", path);
}
std::string DatabaseFilesystem::getTablePath(const std::string & table_name) const
{
fs::path table_path = fs::path(path) / table_name;
return table_path.lexically_normal().string();
}
void DatabaseFilesystem::addTable(const std::string & table_name, StoragePtr table_storage) const
{
std::lock_guard lock(mutex);
auto [_, inserted] = loaded_tables.emplace(table_name, table_storage);
if (!inserted)
throw Exception(
ErrorCodes::LOGICAL_ERROR,
"Table with name `{}` already exists in database `{}` (engine {})",
table_name, getDatabaseName(), getEngineName());
}
bool DatabaseFilesystem::checkTableFilePath(const std::string & table_path, ContextPtr context_, bool throw_on_error) const
{
/// If run in Local mode, no need for path checking.
bool check_path = context_->getApplicationType() != Context::ApplicationType::LOCAL;
const auto & user_files_path = context_->getUserFilesPath();
/// Check access for file before checking its existence.
if (check_path && !fileOrSymlinkPathStartsWith(table_path, user_files_path))
{
if (throw_on_error)
throw Exception(ErrorCodes::PATH_ACCESS_DENIED, "File is not inside {}", user_files_path);
else
return false;
}
/// Check if the corresponding file exists.
if (!fs::exists(table_path))
{
if (throw_on_error)
throw Exception(ErrorCodes::FILE_DOESNT_EXIST, "File does not exist: {}", table_path);
else
return false;
}
if (!fs::is_regular_file(table_path))
{
if (throw_on_error)
throw Exception(ErrorCodes::FILE_DOESNT_EXIST,
"File is directory, but expected a file: {}", table_path);
else
return false;
}
return true;
}
StoragePtr DatabaseFilesystem::tryGetTableFromCache(const std::string & name) const
{
StoragePtr table = nullptr;
{
std::lock_guard lock(mutex);
auto it = loaded_tables.find(name);
if (it != loaded_tables.end())
table = it->second;
}
/// Invalidate cache if file no longer exists.
if (table && !fs::exists(getTablePath(name)))
{
std::lock_guard lock(mutex);
loaded_tables.erase(name);
return nullptr;
}
return table;
}
bool DatabaseFilesystem::isTableExist(const String & name, ContextPtr context_) const
{
if (tryGetTableFromCache(name))
return true;
return checkTableFilePath(getTablePath(name), context_, /* throw_on_error */false);
}
StoragePtr DatabaseFilesystem::getTableImpl(const String & name, ContextPtr context_) const
{
/// Check if table exists in loaded tables map.
if (auto table = tryGetTableFromCache(name))
return table;
auto table_path = getTablePath(name);
checkTableFilePath(table_path, context_, /* throw_on_error */true);
/// If the file exists, create a new table using TableFunctionFile and return it.
auto args = makeASTFunction("file", std::make_shared<ASTLiteral>(table_path));
auto table_function = TableFunctionFactory::instance().get(args, context_);
if (!table_function)
return nullptr;
/// TableFunctionFile throws exceptions, if table cannot be created.
auto table_storage = table_function->execute(args, context_, name);
if (table_storage)
addTable(name, table_storage);
return table_storage;
}
StoragePtr DatabaseFilesystem::getTable(const String & name, ContextPtr context_) const
{
/// getTableImpl can throw exceptions, do not catch them to show correct error to user.
if (auto storage = getTableImpl(name, context_))
return storage;
throw Exception(ErrorCodes::UNKNOWN_TABLE, "Table {}.{} doesn't exist",
backQuoteIfNeed(getDatabaseName()), backQuoteIfNeed(name));
}
StoragePtr DatabaseFilesystem::tryGetTable(const String & name, ContextPtr context_) const
{
try
{
return getTableImpl(name, context_);
}
catch (const Exception & e)
{
/// Ignore exceptions thrown by TableFunctionFile, which indicate that there is no table
/// see tests/02722_database_filesystem.sh for more details.
if (e.code() == ErrorCodes::FILE_DOESNT_EXIST)
{
return nullptr;
}
throw;
}
}
bool DatabaseFilesystem::empty() const
{
std::lock_guard lock(mutex);
return loaded_tables.empty();
}
ASTPtr DatabaseFilesystem::getCreateDatabaseQuery() const
{
const auto & settings = getContext()->getSettingsRef();
const String query = fmt::format("CREATE DATABASE {} ENGINE = Filesystem('{}')", backQuoteIfNeed(getDatabaseName()), path);
ParserCreateQuery parser;
ASTPtr ast = parseQuery(parser, query.data(), query.data() + query.size(), "", 0, settings.max_parser_depth);
if (const auto database_comment = getDatabaseComment(); !database_comment.empty())
{
auto & ast_create_query = ast->as<ASTCreateQuery &>();
ast_create_query.set(ast_create_query.comment, std::make_shared<ASTLiteral>(database_comment));
}
return ast;
}
void DatabaseFilesystem::shutdown()
{
Tables tables_snapshot;
{
std::lock_guard lock(mutex);
tables_snapshot = loaded_tables;
}
for (const auto & kv : tables_snapshot)
{
auto table_id = kv.second->getStorageID();
kv.second->flushAndShutdown();
}
std::lock_guard lock(mutex);
loaded_tables.clear();
}
/**
* Returns an empty vector because the database is read-only and no tables can be backed up
*/
std::vector<std::pair<ASTPtr, StoragePtr>> DatabaseFilesystem::getTablesForBackup(const FilterByNameFunction &, const ContextPtr &) const
{
return {};
}
/**
*
* Returns an empty iterator because the database does not have its own tables
* But only caches them for quick access
*/
DatabaseTablesIteratorPtr DatabaseFilesystem::getTablesIterator(ContextPtr, const FilterByNameFunction &) const
{
return std::make_unique<DatabaseTablesSnapshotIterator>(Tables{}, getDatabaseName());
}
}

67
src/Databases/DatabaseFilesystem.h Normal file
View File

@ -0,0 +1,67 @@
#pragma once
#include <mutex>
#include <Databases/IDatabase.h>
#include <Parsers/IAST.h>
#include <Storages/IStorage_fwd.h>
#include <base/types.h>
namespace DB
{
class Context;
/**
* DatabaseFilesystem allows to interact with files stored on the local filesystem.
* Uses TableFunctionFile to implicitly load file when a user requests the table,
* and provides a read-only access to the data in the file.
* Tables are cached inside the database for quick access
*
* Used in clickhouse-local to access local files.
* For clickhouse-server requires allows to access file only from user_files directory.
*/
class DatabaseFilesystem : public IDatabase, protected WithContext
{
public:
DatabaseFilesystem(const String & name, const String & path, ContextPtr context);
String getEngineName() const override { return "Filesystem"; }
bool isTableExist(const String & name, ContextPtr context) const override;
StoragePtr getTable(const String & name, ContextPtr context) const override;
StoragePtr tryGetTable(const String & name, ContextPtr context) const override;
bool shouldBeEmptyOnDetach() const override { return false; } /// Contains only temporary tables.
bool empty() const override;
bool isReadOnly() const override { return true; }
ASTPtr getCreateDatabaseQuery() const override;
void shutdown() override;
std::vector<std::pair<ASTPtr, StoragePtr>> getTablesForBackup(const FilterByNameFunction &, const ContextPtr &) const override;
DatabaseTablesIteratorPtr getTablesIterator(ContextPtr, const FilterByNameFunction &) const override;
protected:
StoragePtr getTableImpl(const String & name, ContextPtr context) const;
StoragePtr tryGetTableFromCache(const std::string & name) const;
std::string getTablePath(const std::string & table_name) const;
void addTable(const std::string & table_name, StoragePtr table_storage) const;
bool checkTableFilePath(const std::string & table_path, ContextPtr context_, bool throw_on_error) const;
private:
String path;
mutable Tables loaded_tables TSA_GUARDED_BY(mutex);
Poco::Logger * log;
};
}

234
src/Databases/DatabaseHDFS.cpp Normal file
View File

@ -0,0 +1,234 @@
#include "config.h"
#if USE_HDFS
#include <Databases/DatabaseHDFS.h>
#include <Interpreters/Context.h>
#include <Parsers/ASTCreateQuery.h>
#include <Parsers/ASTFunction.h>
#include <Parsers/ASTLiteral.h>
#include <Parsers/parseQuery.h>
#include <Parsers/ParserCreateQuery.h>
#include <Storages/HDFS/HDFSCommon.h>
#include <Storages/IStorage.h>
#include <TableFunctions/TableFunctionFactory.h>
#include <Poco/URI.h>
#include <re2/re2.h>
#include <filesystem>
namespace fs = std::filesystem;
namespace DB
{
namespace ErrorCodes
{
extern const int LOGICAL_ERROR;
extern const int UNKNOWN_TABLE;
extern const int BAD_ARGUMENTS;
extern const int FILE_DOESNT_EXIST;
extern const int UNACCEPTABLE_URL;
extern const int ACCESS_DENIED;
extern const int DATABASE_ACCESS_DENIED;
extern const int HDFS_ERROR;
extern const int CANNOT_EXTRACT_TABLE_STRUCTURE;
}
static constexpr std::string_view HDFS_HOST_REGEXP = "^hdfs://[^/]*";
DatabaseHDFS::DatabaseHDFS(const String & name_, const String & source_url, ContextPtr context_)
: IDatabase(name_)
, WithContext(context_->getGlobalContext())
, source(source_url)
, log(&Poco::Logger::get("DatabaseHDFS(" + name_ + ")"))
{
if (!source.empty())
{
if (!re2::RE2::FullMatch(source, std::string(HDFS_HOST_REGEXP)))
throw Exception(ErrorCodes::BAD_ARGUMENTS, "Bad hdfs host: {}. "
"It should have structure 'hdfs://<host_name>:<port>'", source);
context_->getGlobalContext()->getRemoteHostFilter().checkURL(Poco::URI(source));
}
}
void DatabaseHDFS::addTable(const std::string & table_name, StoragePtr table_storage) const
{
std::lock_guard lock(mutex);
auto [_, inserted] = loaded_tables.emplace(table_name, table_storage);
if (!inserted)
throw Exception(
ErrorCodes::LOGICAL_ERROR,
"Table with name `{}` already exists in database `{}` (engine {})",
table_name, getDatabaseName(), getEngineName());
}
std::string DatabaseHDFS::getTablePath(const std::string & table_name) const
{
if (table_name.starts_with("hdfs://"))
return table_name;
if (source.empty())
throw Exception(ErrorCodes::BAD_ARGUMENTS, "Bad hdfs url: {}. "
"It should have structure 'hdfs://<host_name>:<port>/path'", table_name);
return fs::path(source) / table_name;
}
bool DatabaseHDFS::checkUrl(const std::string & url, ContextPtr context_, bool throw_on_error) const
{
try
{
checkHDFSURL(url);
context_->getGlobalContext()->getRemoteHostFilter().checkURL(Poco::URI(url));
}
catch (...)
{
if (throw_on_error)
throw;
return false;
}
return true;
}
bool DatabaseHDFS::isTableExist(const String & name, ContextPtr context_) const
{
std::lock_guard lock(mutex);
if (loaded_tables.find(name) != loaded_tables.end())
return true;
return checkUrl(name, context_, false);
}
StoragePtr DatabaseHDFS::getTableImpl(const String & name, ContextPtr context_) const
{
/// Check if the table exists in the loaded tables map.
{
std::lock_guard lock(mutex);
auto it = loaded_tables.find(name);
if (it != loaded_tables.end())
return it->second;
}
auto url = getTablePath(name);
checkUrl(url, context_, true);
auto args = makeASTFunction("hdfs", std::make_shared<ASTLiteral>(url));
auto table_function = TableFunctionFactory::instance().get(args, context_);
if (!table_function)
return nullptr;
/// TableFunctionHDFS throws exceptions, if table cannot be created.
auto table_storage = table_function->execute(args, context_, name);
if (table_storage)
addTable(name, table_storage);
return table_storage;
}
StoragePtr DatabaseHDFS::getTable(const String & name, ContextPtr context_) const
{
/// Rethrow all exceptions from TableFunctionHDFS to show correct error to user.
if (auto storage = getTableImpl(name, context_))
return storage;
throw Exception(ErrorCodes::UNKNOWN_TABLE, "Table {}.{} doesn't exist",
backQuoteIfNeed(getDatabaseName()), backQuoteIfNeed(name));
}
StoragePtr DatabaseHDFS::tryGetTable(const String & name, ContextPtr context_) const
{
try
{
return getTableImpl(name, context_);
}
catch (const Exception & e)
{
// Ignore exceptions thrown by TableFunctionHDFS, which indicate that there is no table
if (e.code() == ErrorCodes::BAD_ARGUMENTS
|| e.code() == ErrorCodes::ACCESS_DENIED
|| e.code() == ErrorCodes::DATABASE_ACCESS_DENIED
|| e.code() == ErrorCodes::FILE_DOESNT_EXIST
|| e.code() == ErrorCodes::UNACCEPTABLE_URL
|| e.code() == ErrorCodes::HDFS_ERROR
|| e.code() == ErrorCodes::CANNOT_EXTRACT_TABLE_STRUCTURE)
{
return nullptr;
}
throw;
}
catch (const Poco::URISyntaxException &)
{
return nullptr;
}
}
bool DatabaseHDFS::empty() const
{
std::lock_guard lock(mutex);
return loaded_tables.empty();
}
ASTPtr DatabaseHDFS::getCreateDatabaseQuery() const
{
const auto & settings = getContext()->getSettingsRef();
ParserCreateQuery parser;
const String query = fmt::format("CREATE DATABASE {} ENGINE = HDFS('{}')", backQuoteIfNeed(getDatabaseName()), source);
ASTPtr ast = parseQuery(parser, query.data(), query.data() + query.size(), "", 0, settings.max_parser_depth);
if (const auto database_comment = getDatabaseComment(); !database_comment.empty())
{
auto & ast_create_query = ast->as<ASTCreateQuery &>();
ast_create_query.set(ast_create_query.comment, std::make_shared<ASTLiteral>(database_comment));
}
return ast;
}
void DatabaseHDFS::shutdown()
{
Tables tables_snapshot;
{
std::lock_guard lock(mutex);
tables_snapshot = loaded_tables;
}
for (const auto & kv : tables_snapshot)
{
auto table_id = kv.second->getStorageID();
kv.second->flushAndShutdown();
}
std::lock_guard lock(mutex);
loaded_tables.clear();
}
/**
* Returns an empty vector because the database is read-only and no tables can be backed up
*/
std::vector<std::pair<ASTPtr, StoragePtr>> DatabaseHDFS::getTablesForBackup(const FilterByNameFunction &, const ContextPtr &) const
{
return {};
}
/**
*
* Returns an empty iterator because the database does not have its own tables
* But only caches them for quick access
*/
DatabaseTablesIteratorPtr DatabaseHDFS::getTablesIterator(ContextPtr, const FilterByNameFunction &) const
{
return std::make_unique<DatabaseTablesSnapshotIterator>(Tables{}, getDatabaseName());
}
} // DB
#endif

68
src/Databases/DatabaseHDFS.h Normal file
View File

@ -0,0 +1,68 @@
#pragma once
#include "config.h"
#if USE_HDFS
#include <mutex>
#include <Databases/IDatabase.h>
#include <Parsers/IAST.h>
#include <Storages/IStorage_fwd.h>
#include <base/types.h>
namespace DB
{
class Context;
/**
* DatabaseHDFS allows to interact with files stored on the file system.
* Uses TableFunctionHDFS to implicitly load file when a user requests the table,
* and provides read-only access to the data in the file.
* Tables are cached inside the database for quick access.
*/
class DatabaseHDFS : public IDatabase, protected WithContext
{
public:
DatabaseHDFS(const String & name, const String & source_url, ContextPtr context);
String getEngineName() const override { return "S3"; }
bool isTableExist(const String & name, ContextPtr context) const override;
StoragePtr getTable(const String & name, ContextPtr context) const override;
StoragePtr tryGetTable(const String & name, ContextPtr context) const override;
bool shouldBeEmptyOnDetach() const override { return false; } /// Contains only temporary tables.
bool empty() const override;
bool isReadOnly() const override { return true; }
ASTPtr getCreateDatabaseQuery() const override;
void shutdown() override;
std::vector<std::pair<ASTPtr, StoragePtr>> getTablesForBackup(const FilterByNameFunction &, const ContextPtr &) const override;
DatabaseTablesIteratorPtr getTablesIterator(ContextPtr, const FilterByNameFunction &) const override;
protected:
StoragePtr getTableImpl(const String & name, ContextPtr context) const;
void addTable(const std::string & table_name, StoragePtr table_storage) const;
bool checkUrl(const std::string & url, ContextPtr context_, bool throw_on_error) const;
std::string getTablePath(const std::string & table_name) const;
private:
const String source;
mutable Tables loaded_tables TSA_GUARDED_BY(mutex);
Poco::Logger * log;
};
}
#endif

312
src/Databases/DatabaseS3.cpp Normal file
View File

@ -0,0 +1,312 @@
#include "config.h"
#if USE_AWS_S3
#include <Databases/DatabaseS3.h>
#include <Interpreters/Context.h>
#include <Interpreters/evaluateConstantExpression.h>
#include <IO/S3/URI.h>
#include <Parsers/ASTCreateQuery.h>
#include <Parsers/ASTFunction.h>
#include <Parsers/ASTLiteral.h>
#include <Parsers/parseQuery.h>
#include <Parsers/ParserCreateQuery.h>
#include <Storages/checkAndGetLiteralArgument.h>
#include <Storages/IStorage.h>
#include <Storages/NamedCollectionsHelpers.h>
#include <TableFunctions/TableFunctionFactory.h>
#include <boost/algorithm/string.hpp>
#include <filesystem>
namespace fs = std::filesystem;
namespace DB
{
static const std::unordered_set<std::string_view> optional_configuration_keys = {
"url",
"access_key_id",
"secret_access_key",
"no_sign_request"
};
namespace ErrorCodes
{
extern const int LOGICAL_ERROR;
extern const int UNKNOWN_TABLE;
extern const int BAD_ARGUMENTS;
extern const int FILE_DOESNT_EXIST;
extern const int UNACCEPTABLE_URL;
extern const int S3_ERROR;
extern const int NUMBER_OF_ARGUMENTS_DOESNT_MATCH;
}
DatabaseS3::DatabaseS3(const String & name_, const Configuration& config_, ContextPtr context_)
: IDatabase(name_)
, WithContext(context_->getGlobalContext())
, config(config_)
, log(&Poco::Logger::get("DatabaseS3(" + name_ + ")"))
{
}
void DatabaseS3::addTable(const std::string & table_name, StoragePtr table_storage) const
{
std::lock_guard lock(mutex);
auto [_, inserted] = loaded_tables.emplace(table_name, table_storage);
if (!inserted)
throw Exception(
ErrorCodes::LOGICAL_ERROR,
"Table with name `{}` already exists in database `{}` (engine {})",
table_name, getDatabaseName(), getEngineName());
}
std::string DatabaseS3::getFullUrl(const std::string & name) const
{
if (!config.url_prefix.empty())
return fs::path(config.url_prefix) / name;
return name;
}
bool DatabaseS3::checkUrl(const std::string & url, ContextPtr context_, bool throw_on_error) const
{
try
{
S3::URI uri(url);
context_->getGlobalContext()->getRemoteHostFilter().checkURL(uri.uri);
}
catch (...)
{
if (throw_on_error)
throw;
return false;
}
return true;
}
bool DatabaseS3::isTableExist(const String & name, ContextPtr context_) const
{
std::lock_guard lock(mutex);
if (loaded_tables.find(name) != loaded_tables.end())
return true;
return checkUrl(getFullUrl(name), context_, false);
}
StoragePtr DatabaseS3::getTableImpl(const String & name, ContextPtr context_) const
{
/// Check if the table exists in the loaded tables map.
{
std::lock_guard lock(mutex);
auto it = loaded_tables.find(name);
if (it != loaded_tables.end())
return it->second;
}
auto url = getFullUrl(name);
checkUrl(url, context_, /* throw_on_error */true);
auto function = std::make_shared<ASTFunction>();
function->name = "s3";
function->arguments = std::make_shared<ASTExpressionList>();
function->children.push_back(function->arguments);
function->arguments->children.push_back(std::make_shared<ASTLiteral>(url));
if (config.no_sign_request)
{
function->arguments->children.push_back(std::make_shared<ASTLiteral>("NOSIGN"));
}
else if (config.access_key_id.has_value() && config.secret_access_key.has_value())
{
function->arguments->children.push_back(std::make_shared<ASTLiteral>(config.access_key_id.value()));
function->arguments->children.push_back(std::make_shared<ASTLiteral>(config.secret_access_key.value()));
}
auto table_function = TableFunctionFactory::instance().get(function, context_);
if (!table_function)
return nullptr;
/// TableFunctionS3 throws exceptions, if table cannot be created.
auto table_storage = table_function->execute(function, context_, name);
if (table_storage)
addTable(name, table_storage);
return table_storage;
}
StoragePtr DatabaseS3::getTable(const String & name, ContextPtr context_) const
{
/// Rethrow all exceptions from TableFunctionS3 to show correct error to user.
if (auto storage = getTableImpl(name, context_))
return storage;
throw Exception(ErrorCodes::UNKNOWN_TABLE, "Table {}.{} doesn't exist",
backQuoteIfNeed(getDatabaseName()), backQuoteIfNeed(name));
}
StoragePtr DatabaseS3::tryGetTable(const String & name, ContextPtr context_) const
{
try
{
return getTableImpl(name, context_);
}
catch (const Exception & e)
{
/// Ignore exceptions thrown by TableFunctionS3, which indicate that there is no table.
if (e.code() == ErrorCodes::BAD_ARGUMENTS
|| e.code() == ErrorCodes::S3_ERROR
|| e.code() == ErrorCodes::FILE_DOESNT_EXIST
|| e.code() == ErrorCodes::UNACCEPTABLE_URL)
{
return nullptr;
}
throw;
}
catch (const Poco::URISyntaxException &)
{
return nullptr;
}
}
bool DatabaseS3::empty() const
{
std::lock_guard lock(mutex);
return loaded_tables.empty();
}
ASTPtr DatabaseS3::getCreateDatabaseQuery() const
{
const auto & settings = getContext()->getSettingsRef();
ParserCreateQuery parser;
std::string creation_args;
creation_args += fmt::format("'{}'", config.url_prefix);
if (config.no_sign_request)
creation_args += ", 'NOSIGN'";
else if (config.access_key_id.has_value() && config.secret_access_key.has_value())
creation_args += fmt::format(", '{}', '{}'", config.access_key_id.value(), config.secret_access_key.value());
const String query = fmt::format("CREATE DATABASE {} ENGINE = S3({})", backQuoteIfNeed(getDatabaseName()), creation_args);
ASTPtr ast = parseQuery(parser, query.data(), query.data() + query.size(), "", 0, settings.max_parser_depth);
if (const auto database_comment = getDatabaseComment(); !database_comment.empty())
{
auto & ast_create_query = ast->as<ASTCreateQuery &>();
ast_create_query.set(ast_create_query.comment, std::make_shared<ASTLiteral>(database_comment));
}
return ast;
}
void DatabaseS3::shutdown()
{
Tables tables_snapshot;
{
std::lock_guard lock(mutex);
tables_snapshot = loaded_tables;
}
for (const auto & kv : tables_snapshot)
{
auto table_id = kv.second->getStorageID();
kv.second->flushAndShutdown();
}
std::lock_guard lock(mutex);
loaded_tables.clear();
}
DatabaseS3::Configuration DatabaseS3::parseArguments(ASTs engine_args, ContextPtr context_)
{
Configuration result;
if (auto named_collection = tryGetNamedCollectionWithOverrides(engine_args, context_))
{
auto & collection = *named_collection;
validateNamedCollection(collection, {}, optional_configuration_keys);
result.url_prefix = collection.getOrDefault<String>("url", "");
result.no_sign_request = collection.getOrDefault<bool>("no_sign_request", false);
auto key_id = collection.getOrDefault<String>("access_key_id", "");
auto secret_key = collection.getOrDefault<String>("secret_access_key", "");
if (!key_id.empty())
result.access_key_id = key_id;
if (!secret_key.empty())
result.secret_access_key = secret_key;
}
else
{
const std::string supported_signature =
" - S3()\n"
" - S3('url')\n"
" - S3('url', 'NOSIGN')\n"
" - S3('url', 'access_key_id', 'secret_access_key')\n";
const auto error_message =
fmt::format("Engine DatabaseS3 must have the following arguments signature\n{}", supported_signature);
for (auto & arg : engine_args)
arg = evaluateConstantExpressionOrIdentifierAsLiteral(arg, context_);
if (engine_args.size() > 3)
throw Exception(ErrorCodes::NUMBER_OF_ARGUMENTS_DOESNT_MATCH, error_message.c_str());
if (engine_args.empty())
return result;
result.url_prefix = checkAndGetLiteralArgument<String>(engine_args[0], "url");
// url, NOSIGN
if (engine_args.size() == 2)
{
auto second_arg = checkAndGetLiteralArgument<String>(engine_args[1], "NOSIGN");
if (boost::iequals(second_arg, "NOSIGN"))
result.no_sign_request = true;
else
throw Exception(ErrorCodes::BAD_ARGUMENTS, error_message.c_str());
}
// url, access_key_id, secret_access_key
if (engine_args.size() == 3)
{
auto key_id = checkAndGetLiteralArgument<String>(engine_args[1], "access_key_id");
auto secret_key = checkAndGetLiteralArgument<String>(engine_args[2], "secret_access_key");
if (key_id.empty() || secret_key.empty() || boost::iequals(key_id, "NOSIGN"))
throw Exception(ErrorCodes::BAD_ARGUMENTS, error_message.c_str());
result.access_key_id = key_id;
result.secret_access_key = secret_key;
}
}
return result;
}
/**
* Returns an empty vector because the database is read-only and no tables can be backed up
*/
std::vector<std::pair<ASTPtr, StoragePtr>> DatabaseS3::getTablesForBackup(const FilterByNameFunction &, const ContextPtr &) const
{
return {};
}
/**
*
* Returns an empty iterator because the database does not have its own tables
* But only caches them for quick access
*/
DatabaseTablesIteratorPtr DatabaseS3::getTablesIterator(ContextPtr, const FilterByNameFunction &) const
{
return std::make_unique<DatabaseTablesSnapshotIterator>(Tables{}, getDatabaseName());
}
}
#endif

81
src/Databases/DatabaseS3.h Normal file
View File

@ -0,0 +1,81 @@
#pragma once
#include "config.h"
#if USE_AWS_S3
#include <mutex>
#include <Databases/IDatabase.h>
#include <Parsers/IAST.h>
#include <Storages/IStorage_fwd.h>
#include <base/types.h>
namespace DB
{
class Context;
/**
* DatabaseS3 provides access to data stored in S3.
* Uses TableFunctionS3 to implicitly load file when a user requests the table,
* and provides read-only access to the data in the file.
* Tables are cached inside the database for quick access.
*/
class DatabaseS3 : public IDatabase, protected WithContext
{
public:
struct Configuration
{
std::string url_prefix;
bool no_sign_request = false;
std::optional<std::string> access_key_id;
std::optional<std::string> secret_access_key;
};
DatabaseS3(const String & name, const Configuration& config, ContextPtr context);
String getEngineName() const override { return "S3"; }
bool isTableExist(const String & name, ContextPtr context) const override;
StoragePtr getTable(const String & name, ContextPtr context) const override;
StoragePtr tryGetTable(const String & name, ContextPtr context) const override;
// Contains only temporary tables
bool shouldBeEmptyOnDetach() const override { return false; }
bool empty() const override;
bool isReadOnly() const override { return true; }
ASTPtr getCreateDatabaseQuery() const override;
void shutdown() override;
std::vector<std::pair<ASTPtr, StoragePtr>> getTablesForBackup(const FilterByNameFunction &, const ContextPtr &) const override;
DatabaseTablesIteratorPtr getTablesIterator(ContextPtr, const FilterByNameFunction &) const override;
static Configuration parseArguments(ASTs engine_args, ContextPtr context);
protected:
StoragePtr getTableImpl(const String & name, ContextPtr context) const;
void addTable(const std::string & table_name, StoragePtr table_storage) const;
bool checkUrl(const std::string & url, ContextPtr context_, bool throw_on_error) const;
std::string getFullUrl(const std::string & name) const;
private:
const Configuration config;
mutable Tables loaded_tables TSA_GUARDED_BY(mutex);
Poco::Logger * log;
};
}
#endif

266
src/Databases/DatabasesOverlay.cpp Normal file
View File

@ -0,0 +1,266 @@
#include <Databases/DatabasesOverlay.h>
#include <Common/typeid_cast.h>
#include <Interpreters/Context.h>
#include <Interpreters/InterpreterCreateQuery.h>
#include <Parsers/ASTCreateQuery.h>
#include <Storages/IStorage_fwd.h>
namespace DB
{
namespace ErrorCodes
{
extern const int LOGICAL_ERROR;
extern const int CANNOT_GET_CREATE_TABLE_QUERY;
}
DatabasesOverlay::DatabasesOverlay(const String & name_, ContextPtr context_)
: IDatabase(name_), WithContext(context_->getGlobalContext()), log(&Poco::Logger::get("DatabaseOverlay(" + name_ + ")"))
{
}
DatabasesOverlay & DatabasesOverlay::registerNextDatabase(DatabasePtr database)
{
databases.push_back(std::move(database));
return *this;
}
bool DatabasesOverlay::isTableExist(const String & table_name, ContextPtr context_) const
{
for (const auto & db : databases)
{
if (db->isTableExist(table_name, context_))
return true;
}
return false;
}
StoragePtr DatabasesOverlay::tryGetTable(const String & table_name, ContextPtr context_) const
{
StoragePtr result = nullptr;
for (const auto & db : databases)
{
result = db->tryGetTable(table_name, context_);
if (result)
break;
}
return result;
}
void DatabasesOverlay::createTable(ContextPtr context_, const String & table_name, const StoragePtr & table, const ASTPtr & query)
{
for (auto & db : databases)
{
if (!db->isReadOnly())
{
db->createTable(context_, table_name, table, query);
return;
}
}
throw Exception(
ErrorCodes::LOGICAL_ERROR,
"There is no databases for CREATE TABLE `{}` query in database `{}` (engine {})",
table_name,
getDatabaseName(),
getEngineName());
}
void DatabasesOverlay::dropTable(ContextPtr context_, const String & table_name, bool sync)
{
for (auto & db : databases)
{
if (db->isTableExist(table_name, context_))
{
db->dropTable(context_, table_name, sync);
return;
}
}
throw Exception(
ErrorCodes::LOGICAL_ERROR,
"There is no databases for DROP TABLE `{}` query in database `{}` (engine {})",
table_name,
getDatabaseName(),
getEngineName());
}
void DatabasesOverlay::attachTable(
ContextPtr context_, const String & table_name, const StoragePtr & table, const String & relative_table_path)
{
for (auto & db : databases)
{
try
{
db->attachTable(context_, table_name, table, relative_table_path);
return;
}
catch (...)
{
continue;
}
}
throw Exception(
ErrorCodes::LOGICAL_ERROR,
"There is no databases for ATTACH TABLE `{}` query in database `{}` (engine {})",
table_name,
getDatabaseName(),
getEngineName());
}
StoragePtr DatabasesOverlay::detachTable(ContextPtr context_, const String & table_name)
{
StoragePtr result = nullptr;
for (auto & db : databases)
{
if (db->isTableExist(table_name, context_))
return db->detachTable(context_, table_name);
}
throw Exception(
ErrorCodes::LOGICAL_ERROR,
"There is no databases for DETACH TABLE `{}` query in database `{}` (engine {})",
table_name,
getDatabaseName(),
getEngineName());
}
ASTPtr DatabasesOverlay::getCreateTableQueryImpl(const String & name, ContextPtr context_, bool throw_on_error) const
{
ASTPtr result = nullptr;
for (const auto & db : databases)
{
result = db->tryGetCreateTableQuery(name, context_);
if (result)
break;
}
if (!result && throw_on_error)
throw Exception(
ErrorCodes::CANNOT_GET_CREATE_TABLE_QUERY,
"There is no metadata of table `{}` in database `{}` (engine {})",
name,
getDatabaseName(),
getEngineName());
return result;
}
/*
* DatabaseOverlay cannot be constructed by "CREATE DATABASE" query, as it is not a traditional ClickHouse database
* To use DatabaseOverlay, it must be constructed programmatically in code
*/
ASTPtr DatabasesOverlay::getCreateDatabaseQuery() const
{
return std::make_shared<ASTCreateQuery>();
}
String DatabasesOverlay::getTableDataPath(const String & table_name) const
{
String result;
for (const auto & db : databases)
{
result = db->getTableDataPath(table_name);
if (!result.empty())
break;
}
return result;
}
String DatabasesOverlay::getTableDataPath(const ASTCreateQuery & query) const
{
String result;
for (const auto & db : databases)
{
result = db->getTableDataPath(query);
if (!result.empty())
break;
}
return result;
}
UUID DatabasesOverlay::tryGetTableUUID(const String & table_name) const
{
UUID result = UUIDHelpers::Nil;
for (const auto & db : databases)
{
result = db->tryGetTableUUID(table_name);
if (result != UUIDHelpers::Nil)
break;
}
return result;
}
void DatabasesOverlay::drop(ContextPtr context_)
{
for (auto & db : databases)
db->drop(context_);
}
void DatabasesOverlay::alterTable(ContextPtr local_context, const StorageID & table_id, const StorageInMemoryMetadata & metadata)
{
for (auto & db : databases)
{
if (!db->isReadOnly() && db->isTableExist(table_id.table_name, local_context))
{
db->alterTable(local_context, table_id, metadata);
return;
}
}
throw Exception(
ErrorCodes::LOGICAL_ERROR,
"There is no databases for ALTER TABLE `{}` query in database `{}` (engine {})",
table_id.table_name,
getDatabaseName(),
getEngineName());
}
std::vector<std::pair<ASTPtr, StoragePtr>>
DatabasesOverlay::getTablesForBackup(const FilterByNameFunction & filter, const ContextPtr & local_context) const
{
std::vector<std::pair<ASTPtr, StoragePtr>> result;
for (const auto & db : databases)
{
auto db_backup = db->getTablesForBackup(filter, local_context);
result.insert(result.end(), std::make_move_iterator(db_backup.begin()), std::make_move_iterator(db_backup.end()));
}
return result;
}
void DatabasesOverlay::createTableRestoredFromBackup(
const ASTPtr & create_table_query,
ContextMutablePtr local_context,
std::shared_ptr<IRestoreCoordination> /*restore_coordination*/,
UInt64 /*timeout_ms*/)
{
/// Creates a tables by executing a "CREATE TABLE" query.
InterpreterCreateQuery interpreter{create_table_query, local_context};
interpreter.setInternal(true);
interpreter.execute();
}
bool DatabasesOverlay::empty() const
{
for (const auto & db : databases)
{
if (!db->empty())
return false;
}
return true;
}
void DatabasesOverlay::shutdown()
{
for (auto & db : databases)
db->shutdown();
}
DatabaseTablesIteratorPtr DatabasesOverlay::getTablesIterator(ContextPtr context_, const FilterByNameFunction & filter_by_table_name) const
{
Tables tables;
for (const auto & db : databases)
{
for (auto table_it = db->getTablesIterator(context_, filter_by_table_name); table_it->isValid(); table_it->next())
tables.insert({table_it->name(), table_it->table()});
}
return std::make_unique<DatabaseTablesSnapshotIterator>(std::move(tables), getDatabaseName());
}
}

66
src/Databases/DatabasesOverlay.h Normal file
View File

@ -0,0 +1,66 @@
#pragma once
#include <Storages/IStorage_fwd.h>
#include <Databases/IDatabase.h>
namespace DB
{
/**
* Implements the IDatabase interface and combines multiple other databases
* Searches for tables in each database in order until found, and delegates operations to the appropriate database
* Useful for combining databases
*
* Used in clickhouse-local to combine DatabaseFileSystem and DatabaseMemory
*/
class DatabasesOverlay : public IDatabase, protected WithContext
{
public:
DatabasesOverlay(const String & name_, ContextPtr context_);
/// Not thread-safe. Use only as factory to initialize database
DatabasesOverlay & registerNextDatabase(DatabasePtr database);
String getEngineName() const override { return "Overlay"; }
public:
bool isTableExist(const String & table_name, ContextPtr context) const override;
StoragePtr tryGetTable(const String & table_name, ContextPtr context) const override;
void createTable(ContextPtr context, const String & table_name, const StoragePtr & table, const ASTPtr & query) override;
void dropTable(ContextPtr context, const String & table_name, bool sync) override;
void attachTable(ContextPtr context, const String & table_name, const StoragePtr & table, const String & relative_table_path) override;
StoragePtr detachTable(ContextPtr context, const String & table_name) override;
ASTPtr getCreateTableQueryImpl(const String & name, ContextPtr context, bool throw_on_error) const override;
ASTPtr getCreateDatabaseQuery() const override;
String getTableDataPath(const String & table_name) const override;
String getTableDataPath(const ASTCreateQuery & query) const override;
UUID tryGetTableUUID(const String & table_name) const override;
void drop(ContextPtr context) override;
void alterTable(ContextPtr local_context, const StorageID & table_id, const StorageInMemoryMetadata & metadata) override;
std::vector<std::pair<ASTPtr, StoragePtr>> getTablesForBackup(const FilterByNameFunction & filter, const ContextPtr & local_context) const override;
void createTableRestoredFromBackup(const ASTPtr & create_table_query, ContextMutablePtr local_context, std::shared_ptr<IRestoreCoordination> restore_coordination, UInt64 timeout_ms) override;
DatabaseTablesIteratorPtr getTablesIterator(ContextPtr context, const FilterByNameFunction & filter_by_table_name) const override;
bool empty() const override;
void shutdown() override;
protected:
std::vector<DatabasePtr> databases;
Poco::Logger * log;
};
}

4
src/Databases/IDatabase.h
View File

@ -170,7 +170,7 @@ public:
/// Get the table for work. Return nullptr if there is no table.
virtual StoragePtr tryGetTable(const String & name, ContextPtr context) const = 0;
StoragePtr getTable(const String & name, ContextPtr context) const;
virtual StoragePtr getTable(const String & name, ContextPtr context) const;
virtual UUID tryGetTableUUID(const String & /*table_name*/) const { return UUIDHelpers::Nil; }
@ -183,6 +183,8 @@ public:
/// Is the database empty.
virtual bool empty() const = 0;
virtual bool isReadOnly() const { return false; }
/// Add the table to the database. Record its presence in the metadata.
virtual void createTable(
ContextPtr /*context*/,

7
src/Dictionaries/RegExpTreeDictionary.cpp
View File

@ -30,8 +30,6 @@
#include <Dictionaries/RegExpTreeDictionary.h>
#include <Dictionaries/YAMLRegExpTreeDictionarySource.h>
#include <re2_st/stringpiece.h>
#include "config.h"
#if USE_VECTORSCAN
@ -469,10 +467,9 @@ public:
std::pair<String, bool> processBackRefs(const String & data, const re2_st::RE2 & searcher, const std::vector<StringPiece> & pieces)
{
re2_st::StringPiece haystack(data.data(), data.size());
re2_st::StringPiece matches[10];
std::string_view matches[10];
String result;
searcher.Match(haystack, 0, data.size(), re2_st::RE2::Anchor::UNANCHORED, matches, 10);
searcher.Match({data.data(), data.size()}, 0, data.size(), re2_st::RE2::Anchor::UNANCHORED, matches, 10);
/// if the pattern is a single '$1' but fails to match, we would use the default value.
if (pieces.size() == 1 && pieces[0].ref_num >= 0 && pieces[0].ref_num < 10 && matches[pieces[0].ref_num].empty())
return std::make_pair(result, true);

13
src/Disks/IO/CachedOnDiskReadBufferFromFile.cpp
View File

@ -74,19 +74,22 @@ CachedOnDiskReadBufferFromFile::CachedOnDiskReadBufferFromFile(
}
void CachedOnDiskReadBufferFromFile::appendFilesystemCacheLog(
const FileSegment::Range & file_segment_range, CachedOnDiskReadBufferFromFile::ReadType type)
const FileSegment & file_segment, CachedOnDiskReadBufferFromFile::ReadType type)
{
if (!cache_log)
return;
const auto range = file_segment.range();
FilesystemCacheLogElement elem
{
.event_time = std::chrono::system_clock::to_time_t(std::chrono::system_clock::now()),
.query_id = query_id,
.source_file_path = source_file_path,
.file_segment_range = { file_segment_range.left, file_segment_range.right },
.file_segment_range = { range.left, range.right },
.requested_range = { first_offset, read_until_position },
.file_segment_size = file_segment_range.size(),
.file_segment_key = file_segment.key().toString(),
.file_segment_offset = file_segment.offset(),
.file_segment_size = range.size(),
.read_from_cache_attempted = true,
.read_buffer_id = current_buffer_id,
.profile_counters = std::make_shared<ProfileEvents::Counters::Snapshot>(
@ -495,7 +498,7 @@ bool CachedOnDiskReadBufferFromFile::completeFileSegmentAndGetNext()
auto completed_range = current_file_segment->range();
if (cache_log)
appendFilesystemCacheLog(completed_range, read_type);
appendFilesystemCacheLog(*current_file_segment, read_type);
chassert(file_offset_of_buffer_end > completed_range.right);
@ -518,7 +521,7 @@ CachedOnDiskReadBufferFromFile::~CachedOnDiskReadBufferFromFile()
{
if (cache_log && file_segments && !file_segments->empty())
{
appendFilesystemCacheLog(file_segments->front().range(), read_type);
appendFilesystemCacheLog(file_segments->front(), read_type);
}
}

2
src/Disks/IO/CachedOnDiskReadBufferFromFile.h
View File

@ -90,7 +90,7 @@ private:
bool completeFileSegmentAndGetNext();
void appendFilesystemCacheLog(const FileSegment::Range & file_segment_range, ReadType read_type);
void appendFilesystemCacheLog(const FileSegment & file_segment, ReadType read_type);
bool writeCache(char * data, size_t size, size_t offset, FileSegment & file_segment);

2
src/Disks/IO/ReadBufferFromRemoteFSGather.cpp
View File

@ -90,6 +90,8 @@ void ReadBufferFromRemoteFSGather::appendUncachedReadInfo()
.source_file_path = current_object.remote_path,
.file_segment_range = { 0, current_object.bytes_size },
.cache_type = FilesystemCacheLogElement::CacheType::READ_FROM_FS_BYPASSING_CACHE,
.file_segment_key = {},
.file_segment_offset = {},
.file_segment_size = current_object.bytes_size,
.read_from_cache_attempted = false,
};

4
src/Functions/ReplaceRegexpImpl.h
View File

@ -99,8 +99,8 @@ struct ReplaceRegexpImpl
int num_captures,
const Instructions & instructions)
{
re2_st::StringPiece haystack(haystack_data, haystack_length);
re2_st::StringPiece matches[max_captures];
std::string_view haystack(haystack_data, haystack_length);
std::string_view matches[max_captures];
size_t copy_pos = 0;
size_t match_pos = 0;

8
src/Functions/checkHyperscanRegexp.cpp
View File

@ -45,8 +45,8 @@ bool isLargerThanFifty(std::string_view str)
/// Check for sub-patterns of the form x{n} or x{n,} can be expensive. Ignore spaces before/after n and m.
bool SlowWithHyperscanChecker::isSlowOneRepeat(std::string_view regexp)
{
re2_st::StringPiece haystack(regexp.data(), regexp.size());
re2_st::StringPiece matches[2];
std::string_view haystack(regexp.data(), regexp.size());
std::string_view matches[2];
size_t start_pos = 0;
while (start_pos < haystack.size())
{
@ -67,8 +67,8 @@ bool SlowWithHyperscanChecker::isSlowOneRepeat(std::string_view regexp)
/// Check if sub-patterns of the form x{n,m} can be expensive. Ignore spaces before/after n and m.
bool SlowWithHyperscanChecker::isSlowTwoRepeats(std::string_view regexp)
{
re2_st::StringPiece haystack(regexp.data(), regexp.size());
re2_st::StringPiece matches[3];
std::string_view haystack(regexp.data(), regexp.size());
std::string_view matches[3];
size_t start_pos = 0;
while (start_pos < haystack.size())
{

5
src/Functions/extractAllGroups.h
View File

@ -94,7 +94,6 @@ public:
if (needle.empty())
throw Exception(ErrorCodes::BAD_ARGUMENTS, "Length of 'needle' argument must be greater than 0.");
using StringPiece = typename Regexps::Regexp::StringPieceType;
const Regexps::Regexp holder = Regexps::createRegexp<false, false, false>(needle);
const auto & regexp = holder.getRE2();
@ -111,7 +110,7 @@ public:
groups_count, std::to_string(MAX_GROUPS_COUNT - 1));
// Including 0-group, which is the whole regexp.
PODArrayWithStackMemory<StringPiece, MAX_GROUPS_COUNT> matched_groups(groups_count + 1);
PODArrayWithStackMemory<std::string_view, MAX_GROUPS_COUNT> matched_groups(groups_count + 1);
ColumnArray::ColumnOffsets::MutablePtr root_offsets_col = ColumnArray::ColumnOffsets::create();
ColumnArray::ColumnOffsets::MutablePtr nested_offsets_col = ColumnArray::ColumnOffsets::create();
@ -160,7 +159,7 @@ public:
/// Additional limit to fail fast on supposedly incorrect usage.
const auto max_matches_per_row = context->getSettingsRef().regexp_max_matches_per_row;
PODArray<StringPiece, 0> all_matches;
PODArray<std::string_view, 0> all_matches;
/// Number of times RE matched on each row of haystack column.
PODArray<size_t, 0> number_of_matches_per_row;

4
src/Functions/extractGroups.cpp
View File

@ -75,7 +75,7 @@ public:
throw Exception(ErrorCodes::BAD_ARGUMENTS, "There are no groups in regexp: {}", needle);
// Including 0-group, which is the whole regexp.
PODArrayWithStackMemory<re2_st::StringPiece, 128> matched_groups(groups_count + 1);
PODArrayWithStackMemory<std::string_view, 128> matched_groups(groups_count + 1);
ColumnArray::ColumnOffsets::MutablePtr offsets_col = ColumnArray::ColumnOffsets::create();
ColumnString::MutablePtr data_col = ColumnString::create();
@ -89,7 +89,7 @@ public:
{
std::string_view current_row = column_haystack->getDataAt(i).toView();
if (re2->Match(re2_st::StringPiece(current_row.data(), current_row.size()),
if (re2->Match({current_row.data(), current_row.size()},
0, current_row.size(), re2_st::RE2::UNANCHORED, matched_groups.data(),
static_cast<int>(matched_groups.size())))
{

34
src/Interpreters/Cache/FileCache.cpp
View File

@ -806,6 +806,13 @@ bool FileCache::tryReserve(FileSegment & file_segment, const size_t size)
return true;
}
void FileCache::removeKey(const Key & key)
{
assertInitialized();
auto locked_key = metadata.lockKeyMetadata(key, CacheMetadata::KeyNotFoundPolicy::THROW);
locked_key->removeAll();
}
void FileCache::removeKeyIfExists(const Key & key)
{
assertInitialized();
@ -818,7 +825,14 @@ void FileCache::removeKeyIfExists(const Key & key)
/// But if we have multiple replicated zero-copy tables on the same server
/// it became possible to start removing something from cache when it is used
/// by other "zero-copy" tables. That is why it's not an error.
locked_key->removeAllReleasable();
locked_key->removeAll(/* if_releasable */true);
}
void FileCache::removeFileSegment(const Key & key, size_t offset)
{
assertInitialized();
auto locked_key = metadata.lockKeyMetadata(key, CacheMetadata::KeyNotFoundPolicy::THROW);
locked_key->removeFileSegment(offset);
}
void FileCache::removePathIfExists(const String & path)
@ -830,22 +844,12 @@ void FileCache::removeAllReleasable()
{
assertInitialized();
auto lock = lockCache();
main_priority->iterate([&](LockedKey & locked_key, const FileSegmentMetadataPtr & segment_metadata)
{
if (segment_metadata->releasable())
{
auto file_segment = segment_metadata->file_segment;
locked_key.removeFileSegment(file_segment->offset(), file_segment->lock());
return PriorityIterationResult::REMOVE_AND_CONTINUE;
}
return PriorityIterationResult::CONTINUE;
}, lock);
metadata.iterate([](LockedKey & locked_key) { locked_key.removeAll(/* if_releasable */true); });
if (stash)
{
/// Remove all access information.
auto lock = lockCache();
stash->records.clear();
stash->queue->removeAll(lock);
}
@ -915,7 +919,7 @@ void FileCache::loadMetadata()
continue;
}
const auto key = Key(unhexUInt<UInt128>(key_directory.filename().string().data()));
const auto key = Key::fromKeyString(key_directory.filename().string());
auto locked_key = metadata.lockKeyMetadata(key, CacheMetadata::KeyNotFoundPolicy::CREATE_EMPTY, /* is_initial_load */true);
for (fs::directory_iterator offset_it{key_directory}; offset_it != fs::directory_iterator(); ++offset_it)
@ -1070,7 +1074,7 @@ FileSegmentsHolderPtr FileCache::getSnapshot()
FileSegmentsHolderPtr FileCache::getSnapshot(const Key & key)
{
FileSegments file_segments;
auto locked_key = metadata.lockKeyMetadata(key, CacheMetadata::KeyNotFoundPolicy::THROW);
auto locked_key = metadata.lockKeyMetadata(key, CacheMetadata::KeyNotFoundPolicy::THROW_LOGICAL);
for (const auto & [_, file_segment_metadata] : *locked_key->getKeyMetadata())
file_segments.push_back(FileSegment::getSnapshot(file_segment_metadata->file_segment));
return std::make_unique<FileSegmentsHolder>(std::move(file_segments));

12
src/Interpreters/Cache/FileCache.h
View File

@ -83,13 +83,19 @@ public:
FileSegmentsHolderPtr set(const Key & key, size_t offset, size_t size, const CreateFileSegmentSettings & settings);
/// Remove files by `key`. Removes files which might be used at the moment.
/// Remove file segment by `key` and `offset`. Throws if file segment does not exist.
void removeFileSegment(const Key & key, size_t offset);
/// Remove files by `key`. Throws if key does not exist.
void removeKey(const Key & key);
/// Remove files by `key`.
void removeKeyIfExists(const Key & key);
/// Removes files by `path`. Removes files which might be used at the moment.
/// Removes files by `path`.
void removePathIfExists(const String & path);
/// Remove files by `key`. Will not remove files which are used at the moment.
/// Remove files by `key`.
void removeAllReleasable();
std::vector<String> tryGetCachePaths(const Key & key);

5
src/Interpreters/Cache/FileCacheKey.cpp
View File

@ -28,4 +28,9 @@ FileCacheKey FileCacheKey::random()
return FileCacheKey(UUIDHelpers::generateV4().toUnderType());
}
FileCacheKey FileCacheKey::fromKeyString(const std::string & key_str)
{
return FileCacheKey(unhexUInt<UInt128>(key_str.data()));
}
}

2
src/Interpreters/Cache/FileCacheKey.h
View File

@ -21,6 +21,8 @@ struct FileCacheKey
static FileCacheKey random();
bool operator==(const FileCacheKey & other) const { return key == other.key; }
static FileCacheKey fromKeyString(const std::string & key_str);
};
using FileCacheKeyAndOffset = std::pair<FileCacheKey, size_t>;

67
src/Interpreters/Cache/Metadata.cpp
View File

@ -25,6 +25,7 @@ namespace DB
namespace ErrorCodes
{
extern const int LOGICAL_ERROR;
extern const int BAD_ARGUMENTS;
}
FileSegmentMetadata::FileSegmentMetadata(FileSegmentPtr && file_segment_)
@ -191,6 +192,8 @@ LockedKeyPtr CacheMetadata::lockKeyMetadata(
if (it == end())
{
if (key_not_found_policy == KeyNotFoundPolicy::THROW)
throw Exception(ErrorCodes::BAD_ARGUMENTS, "No such key `{}` in cache", key);
else if (key_not_found_policy == KeyNotFoundPolicy::THROW_LOGICAL)
throw Exception(ErrorCodes::LOGICAL_ERROR, "No such key `{}` in cache", key);
else if (key_not_found_policy == KeyNotFoundPolicy::RETURN_NULL)
return nullptr;
@ -215,6 +218,8 @@ LockedKeyPtr CacheMetadata::lockKeyMetadata(
return locked_metadata;
if (key_not_found_policy == KeyNotFoundPolicy::THROW)
throw Exception(ErrorCodes::BAD_ARGUMENTS, "No such key `{}` in cache", key);
else if (key_not_found_policy == KeyNotFoundPolicy::THROW_LOGICAL)
throw Exception(ErrorCodes::LOGICAL_ERROR, "No such key `{}` in cache", key);
if (key_not_found_policy == KeyNotFoundPolicy::RETURN_NULL)
@ -333,11 +338,11 @@ class DownloadQueue
{
friend struct CacheMetadata;
public:
void add(std::weak_ptr<FileSegment> file_segment)
void add(FileSegmentPtr file_segment)
{
{
std::lock_guard lock(mutex);
queue.push(file_segment);
queue.emplace(file_segment->key(), file_segment->offset(), file_segment);
}
CurrentMetrics::add(CurrentMetrics::FilesystemCacheDownloadQueueElements);
@ -356,8 +361,19 @@ private:
std::mutex mutex;
std::condition_variable cv;
std::queue<std::weak_ptr<FileSegment>> queue;
bool cancelled = false;
struct DownloadInfo
{
CacheMetadata::Key key;
size_t offset;
/// We keep weak pointer to file segment
/// instead of just getting it from file_segment_metadata,
/// because file segment at key:offset count be removed and added back to metadata
/// before we actually started background download.
std::weak_ptr<FileSegment> file_segment;
};
std::queue<DownloadInfo> queue;
};
void CacheMetadata::downloadThreadFunc()
@ -365,6 +381,8 @@ void CacheMetadata::downloadThreadFunc()
std::optional<Memory<>> memory;
while (true)
{
Key key;
size_t offset;
std::weak_ptr<FileSegment> file_segment_weak;
{
@ -379,7 +397,11 @@ void CacheMetadata::downloadThreadFunc()
continue;
}
file_segment_weak = download_queue->queue.front();
auto entry = download_queue->queue.front();
key = entry.key;
offset = entry.offset;
file_segment_weak = entry.file_segment;
download_queue->queue.pop();
}
@ -389,19 +411,21 @@ void CacheMetadata::downloadThreadFunc()
try
{
{
auto file_segment = file_segment_weak.lock();
if (!file_segment
|| file_segment->state() != FileSegment::State::PARTIALLY_DOWNLOADED)
continue;
auto locked_key = lockKeyMetadata(file_segment->key(), KeyNotFoundPolicy::RETURN_NULL);
auto locked_key = lockKeyMetadata(key, KeyNotFoundPolicy::RETURN_NULL);
if (!locked_key)
continue;
auto file_segment_metadata = locked_key->tryGetByOffset(file_segment->offset());
auto file_segment_metadata = locked_key->tryGetByOffset(offset);
if (!file_segment_metadata || file_segment_metadata->evicting())
continue;
auto file_segment = file_segment_weak.lock();
if (!file_segment
|| file_segment != file_segment_metadata->file_segment
|| file_segment->state() != FileSegment::State::PARTIALLY_DOWNLOADED)
continue;
holder = std::make_unique<FileSegmentsHolder>(FileSegments{file_segment});
}
@ -539,11 +563,11 @@ bool LockedKey::isLastOwnerOfFileSegment(size_t offset) const
return file_segment_metadata->file_segment.use_count() == 2;
}
void LockedKey::removeAllReleasable()
void LockedKey::removeAll(bool if_releasable)
{
for (auto it = key_metadata->begin(); it != key_metadata->end();)
{
if (!it->second->releasable())
if (if_releasable && !it->second->releasable())
{
++it;
continue;
@ -564,17 +588,32 @@ void LockedKey::removeAllReleasable()
}
}
KeyMetadata::iterator LockedKey::removeFileSegment(size_t offset)
{
auto it = key_metadata->find(offset);
if (it == key_metadata->end())
throw Exception(ErrorCodes::BAD_ARGUMENTS, "There is no offset {}", offset);
auto file_segment = it->second->file_segment;
return removeFileSegmentImpl(it, file_segment->lock());
}
KeyMetadata::iterator LockedKey::removeFileSegment(size_t offset, const FileSegmentGuard::Lock & segment_lock)
{
auto it = key_metadata->find(offset);
if (it == key_metadata->end())
throw Exception(ErrorCodes::LOGICAL_ERROR, "There is no offset {}", offset);
return removeFileSegmentImpl(it, segment_lock);
}
KeyMetadata::iterator LockedKey::removeFileSegmentImpl(KeyMetadata::iterator it, const FileSegmentGuard::Lock & segment_lock)
{
auto file_segment = it->second->file_segment;
LOG_DEBUG(
key_metadata->log, "Remove from cache. Key: {}, offset: {}, size: {}",
getKey(), offset, file_segment->reserved_size);
getKey(), file_segment->offset(), file_segment->reserved_size);
chassert(file_segment->assertCorrectnessUnlocked(segment_lock));

8
src/Interpreters/Cache/Metadata.h
View File

@ -87,7 +87,7 @@ struct CacheMetadata : public std::unordered_map<FileCacheKey, KeyMetadataPtr>,
{
public:
using Key = FileCacheKey;
using IterateCacheMetadataFunc = std::function<void(const LockedKey &)>;
using IterateCacheMetadataFunc = std::function<void(LockedKey &)>;
explicit CacheMetadata(const std::string & path_);
@ -106,6 +106,7 @@ public:
enum class KeyNotFoundPolicy
{
THROW,
THROW_LOGICAL,
CREATE_EMPTY,
RETURN_NULL,
};
@ -169,9 +170,10 @@ struct LockedKey : private boost::noncopyable
std::shared_ptr<const KeyMetadata> getKeyMetadata() const { return key_metadata; }
std::shared_ptr<KeyMetadata> getKeyMetadata() { return key_metadata; }
void removeAllReleasable();
void removeAll(bool if_releasable = true);
KeyMetadata::iterator removeFileSegment(size_t offset, const FileSegmentGuard::Lock &);
KeyMetadata::iterator removeFileSegment(size_t offset);
void shrinkFileSegmentToDownloadedSize(size_t offset, const FileSegmentGuard::Lock &);
@ -188,6 +190,8 @@ struct LockedKey : private boost::noncopyable
std::string toString() const;
private:
KeyMetadata::iterator removeFileSegmentImpl(KeyMetadata::iterator it, const FileSegmentGuard::Lock &);
const std::shared_ptr<KeyMetadata> key_metadata;
KeyGuard::Lock lock; /// `lock` must be destructed before `key_metadata`.
};

3
src/Interpreters/DatabaseCatalog.cpp
View File

@ -356,7 +356,8 @@ DatabaseAndTable DatabaseCatalog::getTableImpl(
auto table = database->tryGetTable(table_id.table_name, context_);
if (!table && exception)
exception->emplace(Exception(ErrorCodes::UNKNOWN_TABLE, "Table {} doesn't exist", table_id.getNameForLogs()));
exception->emplace(Exception(ErrorCodes::UNKNOWN_TABLE, "Table {} doesn't exist", table_id.getNameForLogs()));
if (!table)
database = nullptr;

4
src/Interpreters/FilesystemCacheLog.cpp
View File

@ -40,6 +40,8 @@ NamesAndTypesList FilesystemCacheLogElement::getNamesAndTypes()
{"source_file_path", std::make_shared<DataTypeString>()},
{"file_segment_range", std::make_shared<DataTypeTuple>(types)},
{"total_requested_range", std::make_shared<DataTypeTuple>(types)},
{"key", std::make_shared<DataTypeString>()},
{"offset", std::make_shared<DataTypeUInt64>()},
{"size", std::make_shared<DataTypeUInt64>()},
{"read_type", std::make_shared<DataTypeString>()},
{"read_from_cache_attempted", std::make_shared<DataTypeUInt8>()},
@ -60,6 +62,8 @@ void FilesystemCacheLogElement::appendToBlock(MutableColumns & columns) const
columns[i++]->insert(source_file_path);
columns[i++]->insert(Tuple{file_segment_range.first, file_segment_range.second});
columns[i++]->insert(Tuple{requested_range.first, requested_range.second});
columns[i++]->insert(file_segment_key);
columns[i++]->insert(file_segment_offset);
columns[i++]->insert(file_segment_size);
columns[i++]->insert(typeToString(cache_type));
columns[i++]->insert(read_from_cache_attempted);

2
src/Interpreters/FilesystemCacheLog.h
View File

@ -39,6 +39,8 @@ struct FilesystemCacheLogElement
std::pair<size_t, size_t> file_segment_range{};
std::pair<size_t, size_t> requested_range{};
CacheType cache_type{};
std::string file_segment_key;
size_t file_segment_offset;
size_t file_segment_size;
bool read_from_cache_attempted;
String read_buffer_id;

13
src/Interpreters/InterpreterSystemQuery.cpp
View File

@ -370,7 +370,18 @@ BlockIO InterpreterSystemQuery::execute()
else
{
auto cache = FileCacheFactory::instance().getByName(query.filesystem_cache_name).cache;
cache->removeAllReleasable();
if (query.delete_key.empty())
{
cache->removeAllReleasable();
}
else
{
auto key = FileCacheKey::fromKeyString(query.delete_key);
if (query.delete_offset.has_value())
cache->removeFileSegment(key, query.delete_offset.value());
else
cache->removeKey(key);
}
}
break;
}

8
src/Parsers/ASTSystemQuery.cpp
View File

@ -210,7 +210,15 @@ void ASTSystemQuery::formatImpl(const FormatSettings & settings, FormatState &,
else if (type == Type::DROP_FILESYSTEM_CACHE)
{
if (!filesystem_cache_name.empty())
{
settings.ostr << (settings.hilite ? hilite_none : "") << " " << filesystem_cache_name;
if (!delete_key.empty())
{
settings.ostr << (settings.hilite ? hilite_none : "") << " KEY " << delete_key;
if (delete_offset.has_value())
settings.ostr << (settings.hilite ? hilite_none : "") << " OFFSET " << delete_offset.value();
}
}
}
else if (type == Type::UNFREEZE)
{

2
src/Parsers/ASTSystemQuery.h
View File

@ -107,6 +107,8 @@ public:
UInt64 seconds{};
String filesystem_cache_name;
std::string delete_key;
std::optional<size_t> delete_offset;
String backup_name;

8
src/Parsers/ParserSystemQuery.cpp
View File

@ -405,7 +405,15 @@ bool ParserSystemQuery::parseImpl(IParser::Pos & pos, ASTPtr & node, Expected &
ParserLiteral path_parser;
ASTPtr ast;
if (path_parser.parse(pos, ast, expected))
{
res->filesystem_cache_name = ast->as<ASTLiteral>()->value.safeGet<String>();
if (ParserKeyword{"KEY"}.ignore(pos, expected) && ParserIdentifier().parse(pos, ast, expected))
{
res->delete_key = ast->as<ASTIdentifier>()->name();
if (ParserKeyword{"OFFSET"}.ignore(pos, expected) && ParserLiteral().parse(pos, ast, expected))
res->delete_offset = ast->as<ASTLiteral>()->value.safeGet<UInt64>();
}
}
if (!parseQueryWithOnCluster(res, pos, expected))
return false;
break;

5
src/Processors/Formats/Impl/RegexpRowInputFormat.h
View File

@ -1,7 +1,6 @@
#pragma once
#include <re2_st/re2.h>
#include <re2_st/stringpiece.h>
#include <string>
#include <vector>
#include <Core/Block.h>
@ -28,14 +27,14 @@ public:
/// Return true if row was successfully parsed and row fields were extracted.
bool parseRow(PeekableReadBuffer & buf);
re2_st::StringPiece getField(size_t index) { return matched_fields[index]; }
std::string_view getField(size_t index) { return matched_fields[index]; }
size_t getMatchedFieldsSize() const { return matched_fields.size(); }
size_t getNumberOfGroups() const { return regexp.NumberOfCapturingGroups(); }
private:
const re2_st::RE2 regexp;
// The vector of fields extracted from line using regexp.
std::vector<re2_st::StringPiece> matched_fields;
std::vector<std::string_view> matched_fields;
// These two vectors are needed to use RE2::FullMatchN (function for extracting fields).
std::vector<re2_st::RE2::Arg> re2_arguments;
std::vector<re2_st::RE2::Arg *> re2_arguments_ptrs;

2
src/Processors/Transforms/buildPushingToViewsChain.cpp
View File

@ -282,7 +282,7 @@ Chain buildPushingToViewsChain(
auto * original_thread = current_thread;
SCOPE_EXIT({ current_thread = original_thread; });
std::unique_ptr<ThreadStatus> view_thread_status_ptr = std::make_unique<ThreadStatus>();
std::unique_ptr<ThreadStatus> view_thread_status_ptr = std::make_unique<ThreadStatus>(/*check_current_thread_on_destruction=*/ false);
/// Copy of a ThreadStatus should be internal.
view_thread_status_ptr->setInternalThread();
view_thread_status_ptr->attachToGroup(running_group);

6
src/Server/HTTPHandler.cpp
View File

@ -44,6 +44,8 @@
#include <Poco/String.h>
#include <Poco/Net/SocketAddress.h>
#include <re2/re2.h>
#include <chrono>
#include <sstream>
@ -1163,8 +1165,8 @@ void PredefinedQueryHandler::customizeContext(HTTPServerRequest & request, Conte
{
int num_captures = compiled_regex->NumberOfCapturingGroups() + 1;
re2::StringPiece matches[num_captures];
re2::StringPiece input(begin, end - begin);
std::string_view matches[num_captures];
std::string_view input(begin, end - begin);
if (compiled_regex->Match(input, 0, end - begin, re2::RE2::Anchor::ANCHOR_BOTH, matches, num_captures))
{
for (const auto & [capturing_name, capturing_index] : compiled_regex->NamedCapturingGroups())

6
src/Server/HTTPHandlerRequestFilter.h
View File

@ -6,7 +6,6 @@
#include <base/find_symbols.h>
#include <re2/re2.h>
#include <re2/stringpiece.h>
#include <Poco/StringTokenizer.h>
#include <Poco/Util/LayeredConfiguration.h>
@ -26,9 +25,8 @@ static inline bool checkRegexExpression(std::string_view match_str, const Compil
{
int num_captures = compiled_regex->NumberOfCapturingGroups() + 1;
re2::StringPiece matches[num_captures];
re2::StringPiece match_input(match_str.data(), match_str.size());
return compiled_regex->Match(match_input, 0, match_str.size(), re2::RE2::Anchor::ANCHOR_BOTH, matches, num_captures);
std::string_view matches[num_captures];
return compiled_regex->Match({match_str.data(), match_str.size()}, 0, match_str.size(), re2::RE2::Anchor::ANCHOR_BOTH, matches, num_captures);
}
static inline bool checkExpression(std::string_view match_str, const std::pair<String, CompiledRegexPtr> & expression)

12
src/Storages/MergeTree/MergeTreeData.cpp
View File

@ -4529,9 +4529,8 @@ MergeTreeData::DataPartPtr MergeTreeData::getActiveContainingPart(
}
void MergeTreeData::swapActivePart(MergeTreeData::DataPartPtr part_copy)
void MergeTreeData::swapActivePart(MergeTreeData::DataPartPtr part_copy, DataPartsLock &)
{
auto lock = lockParts();
for (auto original_active_part : getDataPartsStateRange(DataPartState::Active)) // NOLINT (copy is intended)
{
if (part_copy->name == original_active_part->name)
@ -4587,6 +4586,12 @@ MergeTreeData::DataPartPtr MergeTreeData::getActiveContainingPart(const String &
return getActiveContainingPart(part_info);
}
MergeTreeData::DataPartPtr MergeTreeData::getActiveContainingPart(const String & part_name, DataPartsLock & lock) const
{
auto part_info = MergeTreePartInfo::fromPartName(part_name, format_version);
return getActiveContainingPart(part_info, DataPartState::Active, lock);
}
MergeTreeData::DataPartsVector MergeTreeData::getVisibleDataPartsVectorInPartition(ContextPtr local_context, const String & partition_id) const
{
return getVisibleDataPartsVectorInPartition(local_context->getCurrentTransaction().get(), partition_id);
@ -7191,7 +7196,10 @@ QueryProcessingStage::Enum MergeTreeData::getQueryProcessingStage(
if (query_context->canUseParallelReplicasOnInitiator() && to_stage >= QueryProcessingStage::WithMergeableState)
{
if (!canUseParallelReplicasBasedOnPKAnalysis(query_context, storage_snapshot, query_info))
{
query_info.parallel_replicas_disabled = true;
return QueryProcessingStage::Enum::FetchColumns;
}
/// ReplicatedMergeTree
if (supportsReplication())

3
src/Storages/MergeTree/MergeTreeData.h
View File

@ -504,12 +504,13 @@ public:
/// Returns a part in Active state with the given name or a part containing it. If there is no such part, returns nullptr.
DataPartPtr getActiveContainingPart(const String & part_name) const;
DataPartPtr getActiveContainingPart(const String & part_name, DataPartsLock & lock) const;
DataPartPtr getActiveContainingPart(const MergeTreePartInfo & part_info) const;
DataPartPtr getActiveContainingPart(const MergeTreePartInfo & part_info, DataPartState state, DataPartsLock & lock) const;
/// Swap part with it's identical copy (possible with another path on another disk).
/// If original part is not active or doesn't exist exception will be thrown.
void swapActivePart(MergeTreeData::DataPartPtr part_copy);
void swapActivePart(MergeTreeData::DataPartPtr part_copy, DataPartsLock &);
/// Returns all parts in specified partition
DataPartsVector getVisibleDataPartsVectorInPartition(MergeTreeTransaction * txn, const String & partition_id, DataPartsLock * acquired_lock = nullptr) const;

7
src/Storages/MergeTree/MergeTreePartsMover.cpp
View File

@ -263,7 +263,10 @@ void MergeTreePartsMover::swapClonedPart(TemporaryClonedPart & cloned_part) cons
if (moves_blocker.isCancelled())
throw Exception(ErrorCodes::ABORTED, "Cancelled moving parts.");
auto active_part = data->getActiveContainingPart(cloned_part.part->name);
/// `getActiveContainingPart` and `swapActivePart` are called under the same lock
/// to prevent part becoming inactive between calls
auto part_lock = data->lockParts();
auto active_part = data->getActiveContainingPart(cloned_part.part->name, part_lock);
/// It's ok, because we don't block moving parts for merges or mutations
if (!active_part || active_part->name != cloned_part.part->name)
@ -284,7 +287,7 @@ void MergeTreePartsMover::swapClonedPart(TemporaryClonedPart & cloned_part) cons
cloned_part.part->renameTo(active_part->name, false);
/// TODO what happen if server goes down here?
data->swapActivePart(cloned_part.part);
data->swapActivePart(cloned_part.part, part_lock);
LOG_TRACE(log, "Part {} was moved to {}", cloned_part.part->name, cloned_part.part->getDataPartStorage().getFullPath());

2
src/Storages/SelectQueryInfo.h
View File

@ -255,6 +255,8 @@ struct SelectQueryInfo
Block minmax_count_projection_block;
MergeTreeDataSelectAnalysisResultPtr merge_tree_select_result_ptr;
bool parallel_replicas_disabled = false;
bool is_parameterized_view = false;
NameToNameMap parameterized_view_values;

83
src/Storages/StorageMergeTree.cpp
View File

@ -209,7 +209,9 @@ void StorageMergeTree::read(
size_t max_block_size,
size_t num_streams)
{
if (local_context->canUseParallelReplicasOnInitiator() && local_context->getSettingsRef().parallel_replicas_for_non_replicated_merge_tree)
if (!query_info.parallel_replicas_disabled &&
local_context->canUseParallelReplicasOnInitiator() &&
local_context->getSettingsRef().parallel_replicas_for_non_replicated_merge_tree)
{
auto table_id = getStorageID();
@ -240,7 +242,10 @@ void StorageMergeTree::read(
}
else
{
const bool enable_parallel_reading = local_context->canUseParallelReplicasOnFollower() && local_context->getSettingsRef().parallel_replicas_for_non_replicated_merge_tree;
const bool enable_parallel_reading =
!query_info.parallel_replicas_disabled &&
local_context->canUseParallelReplicasOnFollower() &&
local_context->getSettingsRef().parallel_replicas_for_non_replicated_merge_tree;
if (auto plan = reader.read(
column_names, storage_snapshot, query_info,
@ -929,44 +934,70 @@ MergeMutateSelectedEntryPtr StorageMergeTree::selectPartsToMerge(
SelectPartsDecision select_decision = SelectPartsDecision::CANNOT_SELECT;
if (!canEnqueueBackgroundTask())
auto is_background_memory_usage_ok = [](String * disable_reason) -> bool
{
if (out_disable_reason)
*out_disable_reason = fmt::format("Current background tasks memory usage ({}) is more than the limit ({})",
if (canEnqueueBackgroundTask())
return true;
if (disable_reason)
*disable_reason = fmt::format("Current background tasks memory usage ({}) is more than the limit ({})",
formatReadableSizeWithBinarySuffix(background_memory_tracker.get()),
formatReadableSizeWithBinarySuffix(background_memory_tracker.getSoftLimit()));
}
else if (partition_id.empty())
{
UInt64 max_source_parts_size = merger_mutator.getMaxSourcePartsSizeForMerge();
bool merge_with_ttl_allowed = getTotalMergesWithTTLInMergeList() < data_settings->max_number_of_merges_with_ttl_in_pool;
return false;
};
/// TTL requirements is much more strict than for regular merge, so
/// if regular not possible, than merge with ttl is not also not
/// possible.
if (max_source_parts_size > 0)
if (partition_id.empty())
{
if (is_background_memory_usage_ok(out_disable_reason))
{
select_decision = merger_mutator.selectPartsToMerge(
future_part,
aggressive,
max_source_parts_size,
can_merge,
merge_with_ttl_allowed,
txn,
out_disable_reason);
UInt64 max_source_parts_size = merger_mutator.getMaxSourcePartsSizeForMerge();
bool merge_with_ttl_allowed = getTotalMergesWithTTLInMergeList() < data_settings->max_number_of_merges_with_ttl_in_pool;
/// TTL requirements is much more strict than for regular merge, so
/// if regular not possible, than merge with ttl is not also not
/// possible.
if (max_source_parts_size > 0)
{
select_decision = merger_mutator.selectPartsToMerge(
future_part,
aggressive,
max_source_parts_size,
can_merge,
merge_with_ttl_allowed,
txn,
out_disable_reason);
}
else if (out_disable_reason)
*out_disable_reason = "Current value of max_source_parts_size is zero";
}
else if (out_disable_reason)
*out_disable_reason = "Current value of max_source_parts_size is zero";
}
else
{
while (true)
{
select_decision = merger_mutator.selectAllPartsToMergeWithinPartition(
future_part, can_merge, partition_id, final, metadata_snapshot, txn, out_disable_reason, optimize_skip_merged_partitions);
auto timeout_ms = getSettings()->lock_acquire_timeout_for_background_operations.totalMilliseconds();
auto timeout = std::chrono::milliseconds(timeout_ms);
if (!is_background_memory_usage_ok(out_disable_reason))
{
constexpr auto poll_interval = std::chrono::seconds(1);
Int64 attempts = timeout / poll_interval;
bool ok = false;
for (Int64 i = 0; i < attempts; ++i)
{
std::this_thread::sleep_for(poll_interval);
if (is_background_memory_usage_ok(out_disable_reason))
{
ok = true;
break;
}
}
if (!ok)
break;
}
select_decision = merger_mutator.selectAllPartsToMergeWithinPartition(
future_part, can_merge, partition_id, final, metadata_snapshot, txn, out_disable_reason, optimize_skip_merged_partitions);
/// If final - we will wait for currently processing merges to finish and continue.
if (final
&& select_decision != SelectPartsDecision::SELECTED

2
tests/analyzer_tech_debt.txt
View File

@ -36,6 +36,7 @@
01455_shard_leaf_max_rows_bytes_to_read
01495_subqueries_in_with_statement
01504_rocksdb
01526_client_start_and_exit
01527_dist_sharding_key_dictGet_reload
01528_allow_nondeterministic_optimize_skip_unused_shards
01540_verbatim_partition_pruning
@ -50,6 +51,7 @@
01624_soft_constraints
01651_bugs_from_15889
01656_test_query_log_factories_info
01676_clickhouse_client_autocomplete
01681_bloom_filter_nullable_column
01700_system_zookeeper_path_in
01710_projection_additional_filters

2
tests/ci/stress_tests.lib
View File

@ -243,7 +243,7 @@ function check_logs_for_critical_errors()
# Remove file fatal_messages.txt if it's empty
[ -s /test_output/fatal_messages.txt ] || rm /test_output/fatal_messages.txt
rg -Fa "########################################" /test_output/* > /dev/null \
rg -Faz "########################################" /test_output/* > /dev/null \
&& echo -e "Killed by signal (output files)$FAIL" >> /test_output/test_results.tsv
function get_gdb_log_context()

1
tests/config/config.d/merge_tree_old_dirs_cleanup.xml
View File

@ -5,4 +5,5 @@
<!-- Default is 60 seconds, but let's make tests more aggressive -->
<merge_tree_clear_old_temporary_directories_interval_seconds>5</merge_tree_clear_old_temporary_directories_interval_seconds>
</merge_tree>
<allow_remove_stale_moving_parts>true</allow_remove_stale_moving_parts>
</clickhouse>

5
tests/config/config.d/named_collection.xml
View File

@ -32,5 +32,10 @@
<secret_access_key>testtest</secret_access_key>
<structure>auto</structure>
</s3_conn>
<s3_conn_db>
<url>http://localhost:11111/test/</url>
<access_key_id>test</access_key_id>
<secret_access_key>testtest</secret_access_key>
</s3_conn_db>
</named_collections>
</clickhouse>

45
tests/integration/test_hedged_requests/test.py
View File

@ -203,6 +203,9 @@ def update_configs(
def test_stuck_replica(started_cluster):
if NODES["node"].is_built_with_thread_sanitizer():
pytest.skip("Hedged requests don't work under Thread Sanitizer")
update_configs()
cluster.pause_container("node_1")
@ -233,6 +236,9 @@ def test_stuck_replica(started_cluster):
def test_long_query(started_cluster):
if NODES["node"].is_built_with_thread_sanitizer():
pytest.skip("Hedged requests don't work under Thread Sanitizer")
update_configs()
# Restart to reset pool states.
@ -249,12 +255,18 @@ def test_long_query(started_cluster):
def test_send_table_status_sleep(started_cluster):
if NODES["node"].is_built_with_thread_sanitizer():
pytest.skip("Hedged requests don't work under Thread Sanitizer")
update_configs(node_1_sleep_in_send_tables_status=sleep_time)
check_query(expected_replica="node_2")
check_changing_replica_events(1)
def test_send_table_status_sleep2(started_cluster):
if NODES["node"].is_built_with_thread_sanitizer():
pytest.skip("Hedged requests don't work under Thread Sanitizer")
update_configs(
node_1_sleep_in_send_tables_status=sleep_time,
node_2_sleep_in_send_tables_status=sleep_time,
@ -264,12 +276,18 @@ def test_send_table_status_sleep2(started_cluster):
def test_send_data(started_cluster):
if NODES["node"].is_built_with_thread_sanitizer():
pytest.skip("Hedged requests don't work under Thread Sanitizer")
update_configs(node_1_sleep_in_send_data=sleep_time)
check_query(expected_replica="node_2")
check_changing_replica_events(1)
def test_send_data2(started_cluster):
if NODES["node"].is_built_with_thread_sanitizer():
pytest.skip("Hedged requests don't work under Thread Sanitizer")
update_configs(
node_1_sleep_in_send_data=sleep_time, node_2_sleep_in_send_data=sleep_time
)
@ -278,6 +296,9 @@ def test_send_data2(started_cluster):
def test_combination1(started_cluster):
if NODES["node"].is_built_with_thread_sanitizer():
pytest.skip("Hedged requests don't work under Thread Sanitizer")
update_configs(
node_1_sleep_in_send_tables_status=sleep_time,
node_2_sleep_in_send_data=sleep_time,
@ -287,6 +308,9 @@ def test_combination1(started_cluster):
def test_combination2(started_cluster):
if NODES["node"].is_built_with_thread_sanitizer():
pytest.skip("Hedged requests don't work under Thread Sanitizer")
update_configs(
node_1_sleep_in_send_data=sleep_time,
node_2_sleep_in_send_tables_status=sleep_time,
@ -296,6 +320,9 @@ def test_combination2(started_cluster):
def test_combination3(started_cluster):
if NODES["node"].is_built_with_thread_sanitizer():
pytest.skip("Hedged requests don't work under Thread Sanitizer")
update_configs(
node_1_sleep_in_send_data=sleep_time,
node_2_sleep_in_send_tables_status=1000,
@ -306,6 +333,9 @@ def test_combination3(started_cluster):
def test_combination4(started_cluster):
if NODES["node"].is_built_with_thread_sanitizer():
pytest.skip("Hedged requests don't work under Thread Sanitizer")
update_configs(
node_1_sleep_in_send_tables_status=1000,
node_1_sleep_in_send_data=sleep_time,
@ -317,6 +347,9 @@ def test_combination4(started_cluster):
def test_receive_timeout1(started_cluster):
if NODES["node"].is_built_with_thread_sanitizer():
pytest.skip("Hedged requests don't work under Thread Sanitizer")
# Check the situation when first two replicas get receive timeout
# in establishing connection, but the third replica is ok.
update_configs(
@ -329,6 +362,9 @@ def test_receive_timeout1(started_cluster):
def test_receive_timeout2(started_cluster):
if NODES["node"].is_built_with_thread_sanitizer():
pytest.skip("Hedged requests don't work under Thread Sanitizer")
# Check the situation when first replica get receive timeout
# in packet receiving but there are replicas in process of
# connection establishing.
@ -342,6 +378,9 @@ def test_receive_timeout2(started_cluster):
def test_initial_receive_timeout(started_cluster):
if NODES["node"].is_built_with_thread_sanitizer():
pytest.skip("Hedged requests don't work under Thread Sanitizer")
# Check the situation when replicas don't respond after
# receiving query (so, no packets were send to initiator)
update_configs(
@ -360,6 +399,9 @@ def test_initial_receive_timeout(started_cluster):
def test_async_connect(started_cluster):
if NODES["node"].is_built_with_thread_sanitizer():
pytest.skip("Hedged requests don't work under Thread Sanitizer")
update_configs()
NODES["node"].restart_clickhouse()
@ -390,6 +432,9 @@ def test_async_connect(started_cluster):
def test_async_query_sending(started_cluster):
if NODES["node"].is_built_with_thread_sanitizer():
pytest.skip("Hedged requests don't work under Thread Sanitizer")
update_configs(
node_1_sleep_after_receiving_query=5000,
node_2_sleep_after_receiving_query=5000,

15
tests/integration/test_hedged_requests_parallel/test.py
View File

@ -172,6 +172,9 @@ def update_configs(
def test_send_table_status_sleep(started_cluster):
if NODES["node"].is_built_with_thread_sanitizer():
pytest.skip("Hedged requests don't work under Thread Sanitizer")
update_configs(
node_1_sleep_in_send_tables_status=sleep_time,
node_2_sleep_in_send_tables_status=sleep_time,
@ -181,6 +184,9 @@ def test_send_table_status_sleep(started_cluster):
def test_send_data(started_cluster):
if NODES["node"].is_built_with_thread_sanitizer():
pytest.skip("Hedged requests don't work under Thread Sanitizer")
update_configs(
node_1_sleep_in_send_data=sleep_time, node_2_sleep_in_send_data=sleep_time
)
@ -189,6 +195,9 @@ def test_send_data(started_cluster):
def test_combination1(started_cluster):
if NODES["node"].is_built_with_thread_sanitizer():
pytest.skip("Hedged requests don't work under Thread Sanitizer")
update_configs(
node_1_sleep_in_send_tables_status=1000,
node_2_sleep_in_send_tables_status=1000,
@ -199,6 +208,9 @@ def test_combination1(started_cluster):
def test_combination2(started_cluster):
if NODES["node"].is_built_with_thread_sanitizer():
pytest.skip("Hedged requests don't work under Thread Sanitizer")
update_configs(
node_1_sleep_in_send_data=sleep_time,
node_2_sleep_in_send_tables_status=1000,
@ -210,6 +222,9 @@ def test_combination2(started_cluster):
def test_query_with_no_data_to_sample(started_cluster):
if NODES["node"].is_built_with_thread_sanitizer():
pytest.skip("Hedged requests don't work under Thread Sanitizer")
update_configs(
node_1_sleep_in_send_data=sleep_time, node_2_sleep_in_send_data=sleep_time
)

3
tests/integration/test_secure_socket/test.py
View File

@ -58,6 +58,9 @@ def test(started_cluster):
config.format(sleep_in_send_data_ms=1000000),
)
if NODES["node1"].is_built_with_thread_sanitizer():
pytest.skip("Hedged requests don't work under Thread Sanitizer")
attempts = 0
while attempts < 1000:
setting = NODES["node2"].http_query(

17
tests/integration/test_storage_kerberized_hdfs/hdfs_configs/bootstrap.sh
View File

@ -111,6 +111,23 @@ cat > /usr/local/hadoop/etc/hadoop/hdfs-site.xml << EOF
<name>dfs.datanode.http.address</name>
<value>0.0.0.0:1006</value>
</property>
<!-- If the port is 0 then the server will start on a free port. -->
<property>
<name>dfs.datanode.ipc.address</name>
<value>0.0.0.0:0</value>
</property>
<property>
<name>dfs.namenode.secondary.http-address</name>
<value>0.0.0.0:0</value>
</property>
<property>
<name>dfs.namenode.backup.address</name>
<value>0.0.0.0:0</value>
</property>
<property>
<name>dfs.namenode.backup.http-address</name>
<value>0.0.0.0:0</value>
</property>
<!--
<property>
<name>dfs.http.policy</name>

45
tests/performance/join_set_filter.xml
View File

@ -1,45 +0,0 @@
<test>
<substitutions>
<substitution>
<name>table_size</name>
<values>
<value>100000000</value>
</values>
</substitution>
</substitutions>
<settings>
<join_algorithm>full_sorting_merge</join_algorithm>
</settings>
<create_query>
CREATE TABLE t1 (x UInt64, y UInt64) ENGINE = MergeTree ORDER BY y
AS SELECT
sipHash64(number, 't1_x') % {table_size} AS x,
sipHash64(number, 't1_y') % {table_size} AS y
FROM numbers({table_size})
</create_query>
<create_query>
CREATE TABLE t2 (x UInt64, y UInt64) ENGINE = MergeTree ORDER BY y
AS SELECT
sipHash64(number, 't2_x') % {table_size} AS x,
sipHash64(number, 't2_y') % {table_size} AS y
FROM numbers({table_size})
</create_query>
<query>SELECT * FROM t1 JOIN t2 ON t1.x = t2.x WHERE less(t1.y, 10000)</query>
<query>SELECT * FROM t2 JOIN t1 ON t1.x = t2.x WHERE less(t1.y, 10000)</query>
<query>SELECT * FROM t1 JOIN t2 ON t1.x = t2.x WHERE greater(t1.y, {table_size} - 10000)</query>
<query>SELECT * FROM t2 JOIN t1 ON t1.x = t2.x WHERE greater(t1.y, {table_size} - 10000)</query>
<query>SELECT * FROM t1 JOIN t2 ON t1.x = t2.x WHERE t1.y % 100 = 0</query>
<query>SELECT * FROM t2 JOIN t1 ON t1.x = t2.x WHERE t1.y % 100 = 0</query>
<query>SELECT * FROM t1 JOIN t2 ON t1.x = t2.x WHERE t1.y % 1000 = 0</query>
<query>SELECT * FROM t2 JOIN t1 ON t1.x = t2.x WHERE t1.y % 1000 = 0</query>
<drop_query>DROP TABLE IF EXISTS t1</drop_query>
<drop_query>DROP TABLE IF EXISTS t2</drop_query>
</test>

2
tests/queries/0_stateless/01361_fover_remote_num_tries.sh
View File

@ -5,4 +5,4 @@ CURDIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)
# shellcheck source=../shell_config.sh
. "$CURDIR"/../shell_config.sh
$CLICKHOUSE_CLIENT --connections_with_failover_max_tries 10 --query "SELECT hostName() FROM remote('128.1.2.3', default.tmp)" 2>&1 | grep -o -P 'Timeout exceeded while connecting to socket|Network is unreachable' | wc -l
$CLICKHOUSE_CLIENT --connections_with_failover_max_tries 10 --query "SELECT hostName() FROM remote('128.1.2.3', default.tmp)" 2>&1 | grep -o -P 'Timeout exceeded while connecting to socket|Network is unreachable|Timeout: connect timed out' | wc -l

1
tests/queries/0_stateless/01851_hedged_connections_external_tables.sql
View File

@ -1 +1,2 @@
-- Tags: no-tsan
select number from remote('127.0.0.{3|2}', numbers(2)) where number global in (select number from numbers(1))　settings async_socket_for_remote=1, use_hedged_requests = 1, sleep_in_send_data_ms=10, receive_data_timeout_ms=1;

8
tests/queries/0_stateless/01956_skip_unavailable_shards_excessive_attempts.reference
View File

@ -1,10 +1,10 @@
255.255.255.255
HedgedConnectionsFactory: Connection failed at try №1
ConnectionPoolWithFailover: Connection failed at try №1
executeQuery: Code: 519.: All attempts to get table structure failed.
127.2,255.255.255.255
0
HedgedConnectionsFactory: Connection failed at try №1
ConnectionPoolWithFailover: Connection failed at try №1
255.255.255.255,127.2
0
HedgedConnectionsFactory: Connection failed at try №1
HedgedConnectionsFactory: Connection failed at try №1
ConnectionPoolWithFailover: Connection failed at try №1
ConnectionPoolWithFailover: Connection failed at try №1

2
tests/queries/0_stateless/01956_skip_unavailable_shards_excessive_attempts.sh
View File

@ -25,7 +25,7 @@ function execute_query()
# clickhouse-client 2> >(wc -l)
#
# May dump output of "wc -l" after some other programs.
$CLICKHOUSE_CLIENT "${opts[@]}" --query "select * from remote('$hosts', system.one)" 2>"$stderr"
$CLICKHOUSE_CLIENT "${opts[@]}" --query "select * from remote('$hosts', system.one) settings use_hedged_requests=0" 2>"$stderr"
process_log_safe "$stderr"
}
execute_query 255.255.255.255

2
tests/queries/0_stateless/02226_parallel_reading_from_replicas_benchmark.sh
View File

@ -17,6 +17,8 @@ opts=(
--allow_experimental_parallel_reading_from_replicas 1
--parallel_replicas_for_non_replicated_merge_tree 1
--max_parallel_replicas 3
--use_hedged_requests 0
--cluster_for_parallel_replicas parallel_replicas
--iterations 1
)

2
tests/queries/0_stateless/02302_lc_nullable_string_insert_as_number.reference Normal file
View File

@ -0,0 +1,2 @@
0
1

11
tests/queries/0_stateless/02302_lc_nullable_string_insert_as_number.sql Normal file
View File

@ -0,0 +1,11 @@
DROP TABLE IF EXISTS lc_nullable_string;
CREATE TABLE lc_nullable_string(`c1` LowCardinality(Nullable(String)) DEFAULT CAST(NULL, 'LowCardinality(Nullable(String))'))
ENGINE = Memory;
INSERT INTO lc_nullable_string (c1) FORMAT Values (0);
INSERT INTO lc_nullable_string (c1) Values (1);
SELECT * FROM lc_nullable_string ORDER BY c1;
DROP TABLE lc_nullable_string;

4
tests/queries/0_stateless/02581_share_big_sets_between_multiple_mutations_tasks_long.sql
View File

@ -1,4 +1,6 @@
-- Tags: long, no-debug, no-tsan, no-asan, no-ubsan, no-msan
-- Tags: long, no-debug, no-tsan, no-asan, no-ubsan, no-msan, no-parallel
-- no-parallel because the sets use a lot of memory, which may interfere with other tests
DROP TABLE IF EXISTS 02581_trips;

4
tests/queries/0_stateless/02581_share_big_sets_between_mutation_tasks_long.sql
View File

@ -1,4 +1,6 @@
-- Tags: long, no-debug, no-tsan, no-asan, no-ubsan, no-msan
-- Tags: long, no-debug, no-tsan, no-asan, no-ubsan, no-msan, no-parallel
-- no-parallel because the sets use a lot of memory, which may interfere with other tests
DROP TABLE IF EXISTS 02581_trips;

9
tests/queries/0_stateless/02707_clickhouse_local_implicit_file_table_function.reference Normal file
View File

@ -0,0 +1,9 @@
Test 1: check explicit and implicit call of the file table function
explicit:
4
implicit:
4
Test 2: check Filesystem database
4
Test 3: check show database with Filesystem
test02707

45
tests/queries/0_stateless/02707_clickhouse_local_implicit_file_table_function.sh Executable file
View File

@ -0,0 +1,45 @@
#!/usr/bin/env bash
CURDIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)
# shellcheck source=../shell_config.sh
. "$CURDIR"/../shell_config.sh
dir=${CLICKHOUSE_TEST_UNIQUE_NAME}
[[ -d $dir ]] && rm -rd $dir
mkdir $dir
# Create temporary csv file for tests
echo '"id","str","int","text"' > $dir/tmp.csv
echo '1,"abc",123,"abacaba"' >> $dir/tmp.csv
echo '2,"def",456,"bacabaa"' >> $dir/tmp.csv
echo '3,"story",78912,"acabaab"' >> $dir/tmp.csv
echo '4,"history",21321321,"cabaaba"' >> $dir/tmp.csv
#################
echo "Test 1: check explicit and implicit call of the file table function"
echo "explicit:"
$CLICKHOUSE_LOCAL -q "SELECT COUNT(*) FROM file('${dir}/tmp.csv')"
echo "implicit:"
$CLICKHOUSE_LOCAL -q "SELECT COUNT(*) FROM \"${dir}/tmp.csv\""
#################
echo "Test 2: check Filesystem database"
$CLICKHOUSE_LOCAL --multiline --multiquery -q """
DROP DATABASE IF EXISTS test;
CREATE DATABASE test ENGINE = Filesystem('${dir}');
SELECT COUNT(*) FROM test.\`tmp.csv\`;
DROP DATABASE test;
"""
#################
echo "Test 3: check show database with Filesystem"
$CLICKHOUSE_LOCAL --multiline --multiquery -q """
DROP DATABASE IF EXISTS test02707;
CREATE DATABASE test02707 ENGINE = Filesystem('${dir}');
SHOW DATABASES;
DROP DATABASE test02707;
""" | grep "test02707"
# Remove temporary dir with files
rm -rd $dir

15
tests/queries/0_stateless/02722_database_filesystem.reference Normal file
View File

@ -0,0 +1,15 @@
Test 1: create filesystem database and check implicit calls
0
test1
4
4
4
Test 2: check DatabaseFilesystem access rights and errors handling on server
OK
OK
OK
OK
OK
OK
OK
OK

72
tests/queries/0_stateless/02722_database_filesystem.sh Executable file
View File

@ -0,0 +1,72 @@
#!/usr/bin/env bash
# Tags: no-parallel
CURDIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)
# shellcheck source=../shell_config.sh
. "$CURDIR"/../shell_config.sh
# see 01658_read_file_to_stringcolumn.sh
CLICKHOUSE_USER_FILES_PATH=$(clickhouse-client --query "select _path, _file from file('nonexist.txt', 'CSV', 'val1 char')" 2>&1 | grep Exception | awk '{gsub("/nonexist.txt","",$9); print $9}')
# Prepare data
unique_name=${CLICKHOUSE_TEST_UNIQUE_NAME}
user_files_tmp_dir=${CLICKHOUSE_USER_FILES_PATH}/${unique_name}
mkdir -p ${user_files_tmp_dir}/tmp/
echo '"id","str","int","text"' > ${user_files_tmp_dir}/tmp.csv
echo '1,"abc",123,"abacaba"' >> ${user_files_tmp_dir}/tmp.csv
echo '2,"def",456,"bacabaa"' >> ${user_files_tmp_dir}/tmp.csv
echo '3,"story",78912,"acabaab"' >> ${user_files_tmp_dir}/tmp.csv
echo '4,"history",21321321,"cabaaba"' >> ${user_files_tmp_dir}/tmp.csv
tmp_dir=${CLICKHOUSE_TEST_UNIQUE_NAME}
[[ -d $tmp_dir ]] && rm -rd $tmp_dir
mkdir $tmp_dir
cp ${user_files_tmp_dir}/tmp.csv ${tmp_dir}/tmp.csv
cp ${user_files_tmp_dir}/tmp.csv ${user_files_tmp_dir}/tmp/tmp.csv
cp ${user_files_tmp_dir}/tmp.csv ${user_files_tmp_dir}/tmp.myext
#################
echo "Test 1: create filesystem database and check implicit calls"
${CLICKHOUSE_CLIENT} --multiline --multiquery -q """
DROP DATABASE IF EXISTS test1;
CREATE DATABASE test1 ENGINE = Filesystem;
"""
echo $?
${CLICKHOUSE_CLIENT} --query "SHOW DATABASES" | grep "test1"
${CLICKHOUSE_CLIENT} --query "SELECT COUNT(*) FROM test1.\`${unique_name}/tmp.csv\`;"
${CLICKHOUSE_CLIENT} --query "SELECT COUNT(*) FROM test1.\`${unique_name}/tmp/tmp.csv\`;"
${CLICKHOUSE_LOCAL} -q "SELECT COUNT(*) FROM \"${tmp_dir}/tmp.csv\""
#################
echo "Test 2: check DatabaseFilesystem access rights and errors handling on server"
# DATABASE_ACCESS_DENIED: Allows list files only inside user_files
${CLICKHOUSE_CLIENT} --query "SELECT COUNT(*) FROM test1.\`../tmp.csv\`;" 2>&1| grep -F "Code: 481" > /dev/null && echo "OK" || echo 'FAIL' ||:
${CLICKHOUSE_CLIENT} --query "SELECT COUNT(*) FROM test1.\`/tmp/tmp.csv\`;" 2>&1| grep -F "Code: 481" > /dev/null && echo "OK" || echo 'FAIL' ||:
${CLICKHOUSE_CLIENT} --multiline --multiquery --query """
USE test1;
SELECT COUNT(*) FROM \"../${tmp_dir}/tmp.csv\";
""" 2>&1| grep -F "Code: 481" > /dev/null && echo "OK" || echo 'FAIL' ||:
${CLICKHOUSE_CLIENT} --query "SELECT COUNT(*) FROM test1.\`../../../../../../tmp.csv\`;" 2>&1| grep -F "Code: 481" > /dev/null && echo "OK" || echo 'FAIL' ||:
# BAD_ARGUMENTS: path should be inside user_files
${CLICKHOUSE_CLIENT} --multiline --multiquery -q """
DROP DATABASE IF EXISTS test2;
CREATE DATABASE test2 ENGINE = Filesystem('/tmp');
""" 2>&1| grep -F "Code: 36" > /dev/null && echo "OK" || echo 'FAIL' ||:
# BAD_ARGUMENTS: .../user_files/relative_unknown_dir does not exists
${CLICKHOUSE_CLIENT} --multiline --multiquery -q """
DROP DATABASE IF EXISTS test2;
CREATE DATABASE test2 ENGINE = Filesystem('relative_unknown_dir');
""" 2>&1| grep -F "Code: 36" > /dev/null && echo "OK" || echo 'FAIL' ||:
# FILE_DOESNT_EXIST: unknown file
${CLICKHOUSE_CLIENT} --query "SELECT COUNT(*) FROM test1.\`tmp2.csv\`;" 2>&1| grep -F "Code: 60" > /dev/null && echo "OK" || echo 'FAIL' ||:
# BAD_ARGUMENTS: Cannot determine the file format by it's extension
${CLICKHOUSE_CLIENT} --query "SELECT COUNT(*) FROM test1.\`${unique_name}/tmp.myext\`;" 2>&1| grep -F "Code: 36" > /dev/null && echo "OK" || echo 'FAIL' ||:
# Clean
${CLICKHOUSE_CLIENT} --query "DROP DATABASE test1;"
rm -rd $tmp_dir
rm -rd $user_files_tmp_dir

21
tests/queries/0_stateless/02724_database_s3.reference Normal file
View File

@ -0,0 +1,21 @@
Test 1: select from s3
1 2 3
4 5 6
7 8 9
0 0 0
test1
10 11 12
13 14 15
16 17 18
0 0 0
10 11 12
13 14 15
16 17 18
0 0 0
10 11 12
13 14 15
16 17 18
0 0 0
Test 2: check exceptions
OK
OK

63
tests/queries/0_stateless/02724_database_s3.sh Executable file
View File

@ -0,0 +1,63 @@
#!/usr/bin/env bash
# Tags: no-fasttest, no-parallel
# Tag no-fasttest: Depends on AWS
CUR_DIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)
# shellcheck source=../shell_config.sh
. "$CUR_DIR"/../shell_config.sh
#################
echo "Test 1: select from s3"
${CLICKHOUSE_CLIENT} --multiline --multiquery -q """
DROP DATABASE IF EXISTS test1;
CREATE DATABASE test1 ENGINE = S3;
USE test1;
SELECT * FROM \"http://localhost:11111/test/a.tsv\"
"""
${CLICKHOUSE_CLIENT} -q "SHOW DATABASES;" | grep test1
# check credentials with absolute path
${CLICKHOUSE_CLIENT} --multiline --multiquery -q """
DROP DATABASE IF EXISTS test2;
CREATE DATABASE test2 ENGINE = S3('', 'test', 'testtest');
USE test2;
SELECT * FROM \"http://localhost:11111/test/b.tsv\"
"""
# check credentials with relative path
${CLICKHOUSE_CLIENT} --multiline --multiquery -q """
DROP DATABASE IF EXISTS test4;
CREATE DATABASE test4 ENGINE = S3('http://localhost:11111/test', 'test', 'testtest');
USE test4;
SELECT * FROM \"b.tsv\"
"""
# Check named collection loading
${CLICKHOUSE_CLIENT} --multiline --multiquery -q """
DROP DATABASE IF EXISTS test5;
CREATE DATABASE test5 ENGINE = S3(s3_conn_db);
SELECT * FROM test5.\`b.tsv\`
"""
#################
echo "Test 2: check exceptions"
${CLICKHOUSE_CLIENT} --multiline --multiquery -q """
DROP DATABASE IF EXISTS test3;
CREATE DATABASE test3 ENGINE = S3;
USE test3;
SELECT * FROM \"http://localhost:11111/test/a.myext\"
""" 2>&1| grep -F "UNKNOWN_TABLE" > /dev/null && echo "OK"
${CLICKHOUSE_CLIENT} --multiline --multiquery -q """
USE test3;
SELECT * FROM \"abacaba\"
""" 2>&1| grep -F "UNKNOWN_TABLE" > /dev/null && echo "OK"
# Cleanup
${CLICKHOUSE_CLIENT} --multiline --multiquery -q """
DROP DATABASE IF EXISTS test1;
DROP DATABASE IF EXISTS test2;
DROP DATABASE IF EXISTS test3;
DROP DATABASE IF EXISTS test4;
DROP DATABASE IF EXISTS test5;
"""

12
tests/queries/0_stateless/02725_database_hdfs.reference Normal file
View File

@ -0,0 +1,12 @@
Test 1: select from hdfs database
1 2 3
test1
1 2 3
test2
Test 2: check exceptions
OK0
OK1
OK2
OK3
OK4
OK5

60
tests/queries/0_stateless/02725_database_hdfs.sh Executable file
View File

@ -0,0 +1,60 @@
#!/usr/bin/env bash
# Tags: no-fasttest, use-hdfs, no-parallel
CURDIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)
# shellcheck source=../shell_config.sh
. "$CURDIR"/../shell_config.sh
# Prepare data
${CLICKHOUSE_CLIENT} -q "insert into table function hdfs('hdfs://localhost:12222/test_02725_1.tsv', 'TSV', 'column1 UInt32, column2 UInt32, column3 UInt32') select 1, 2, 3 settings hdfs_truncate_on_insert=1;"
${CLICKHOUSE_CLIENT} -q "insert into table function hdfs('hdfs://localhost:12222/test_02725_2.tsv', 'TSV', 'column1 UInt32, column2 UInt32, column3 UInt32') select 4, 5, 6 settings hdfs_truncate_on_insert=1;"
#################
echo "Test 1: select from hdfs database"
# Database without specific host
${CLICKHOUSE_CLIENT} --multiline --multiquery -q """
DROP DATABASE IF EXISTS test1;
CREATE DATABASE test1 ENGINE = HDFS;
USE test1;
SELECT * FROM \"hdfs://localhost:12222/test_02725_1.tsv\"
"""
${CLICKHOUSE_CLIENT} -q "SHOW DATABASES;" | grep test1
# Database with host
${CLICKHOUSE_CLIENT} --multiline --multiquery -q """
DROP DATABASE IF EXISTS test2;
CREATE DATABASE test2 ENGINE = HDFS('hdfs://localhost:12222');
USE test2;
SELECT * FROM \"test_02725_1.tsv\"
"""
${CLICKHOUSE_CLIENT} -q "SHOW DATABASES;" | grep test2
#################
echo "Test 2: check exceptions"
${CLICKHOUSE_CLIENT} --multiline --multiquery -q """
DROP DATABASE IF EXISTS test3;
CREATE DATABASE test3 ENGINE = HDFS('abacaba');
""" 2>&1| grep -F "BAD_ARGUMENTS" > /dev/null && echo "OK0"
${CLICKHOUSE_CLIENT} --multiline --multiquery -q """
DROP DATABASE IF EXISTS test4;
CREATE DATABASE test4 ENGINE = HDFS;
USE test4;
SELECT * FROM \"abacaba/file.tsv\"
""" 2>&1| grep -F "UNKNOWN_TABLE" > /dev/null && echo "OK1"
${CLICKHOUSE_CLIENT} -q "SELECT * FROM test4.\`http://localhost:11111/test/a.tsv\`" 2>&1| grep -F "UNKNOWN_TABLE" > /dev/null && echo "OK2"
${CLICKHOUSE_CLIENT} --query "SELECT * FROM test4.\`hdfs://localhost:12222/file.myext\`" 2>&1| grep -F "UNKNOWN_TABLE" > /dev/null && echo "OK3"
${CLICKHOUSE_CLIENT} --query "SELECT * FROM test4.\`hdfs://localhost:12222/test_02725_3.tsv\`" 2>&1| grep -F "UNKNOWN_TABLE" > /dev/null && echo "OK4"
${CLICKHOUSE_CLIENT} --query "SELECT * FROM test4.\`hdfs://localhost:12222\`" 2>&1| grep -F "UNKNOWN_TABLE" > /dev/null && echo "OK5"
# Cleanup
${CLICKHOUSE_CLIENT} --multiline --multiquery -q """
DROP DATABASE IF EXISTS test1;
DROP DATABASE IF EXISTS test2;
DROP DATABASE IF EXISTS test3;
DROP DATABASE IF EXISTS test4;
"""

4
tests/queries/0_stateless/02808_filesystem_cache_drop_query.reference Normal file
View File

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

67
tests/queries/0_stateless/02808_filesystem_cache_drop_query.sh Executable file
View File

@ -0,0 +1,67 @@
#!/usr/bin/env bash
# Tags: no-fasttest, no-parallel, no-s3-storage, no-random-settings
# set -x
CUR_DIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)
# shellcheck source=../shell_config.sh
. "$CUR_DIR"/../shell_config.sh
disk_name="${CLICKHOUSE_TEST_UNIQUE_NAME}"
$CLICKHOUSE_CLIENT -nm --query """
DROP TABLE IF EXISTS test;
CREATE TABLE test (a Int32, b String)
ENGINE = MergeTree() ORDER BY tuple()
SETTINGS disk = disk_$disk_name(type = cache, max_size = '100Ki', path = ${CLICKHOUSE_TEST_UNIQUE_NAME}, disk = s3_disk);
INSERT INTO test SELECT 1, 'test';
"""
query_id=$RANDOM
$CLICKHOUSE_CLIENT --query_id "$query_id" --query "SELECT * FROM test FORMAT Null SETTINGS enable_filesystem_cache_log = 1"
${CLICKHOUSE_CLIENT} -q " system flush logs"
key=$($CLICKHOUSE_CLIENT -nm --query """
SELECT key FROM system.filesystem_cache_log WHERE query_id = '$query_id' ORDER BY size DESC LIMIT 1;
""")
offset=$($CLICKHOUSE_CLIENT -nm --query """
SELECT offset FROM system.filesystem_cache_log WHERE query_id = '$query_id' ORDER BY size DESC LIMIT 1;
""")
$CLICKHOUSE_CLIENT -nm --query """
SELECT count() FROM system.filesystem_cache WHERE key = '$key' AND file_segment_range_begin = $offset;
"""
$CLICKHOUSE_CLIENT -nm --query """
SYSTEM DROP FILESYSTEM CACHE '$disk_name' KEY $key OFFSET $offset;
"""
$CLICKHOUSE_CLIENT -nm --query """
SELECT count() FROM system.filesystem_cache WHERE key = '$key' AND file_segment_range_begin = $offset;
"""
query_id=$RANDOM$RANDOM
$CLICKHOUSE_CLIENT --query_id "$query_id" --query "SELECT * FROM test FORMAT Null SETTINGS enable_filesystem_cache_log = 1"
${CLICKHOUSE_CLIENT} -q " system flush logs"
key=$($CLICKHOUSE_CLIENT -nm --query """
SELECT key FROM system.filesystem_cache_log WHERE query_id = '$query_id' ORDER BY size DESC LIMIT 1;
""")
$CLICKHOUSE_CLIENT -nm --query """
SELECT count() FROM system.filesystem_cache WHERE key = '$key';
"""
$CLICKHOUSE_CLIENT -nm --query """
SYSTEM DROP FILESYSTEM CACHE '$disk_name' KEY $key
"""
$CLICKHOUSE_CLIENT -nm --query """
SELECT count() FROM system.filesystem_cache WHERE key = '$key';
"""

4
tests/queries/0_stateless/02811_parallel_replicas_prewhere_count.reference Normal file
View File

@ -0,0 +1,4 @@
-- count() ------------------------------
2
-- count() with parallel replicas -------
2

24
tests/queries/0_stateless/02811_parallel_replicas_prewhere_count.sql Normal file
View File

@ -0,0 +1,24 @@
DROP TABLE IF EXISTS users;
CREATE TABLE users (uid Int16, name String, age Int16) ENGINE=MergeTree() ORDER BY uid;
INSERT INTO users VALUES (111, 'JFK', 33);
INSERT INTO users VALUES (6666, 'KLM', 48);
INSERT INTO users VALUES (88888, 'AMS', 50);
SELECT '-- count() ------------------------------';
SELECT count() FROM users PREWHERE uid > 2000;
-- enable parallel replicas but with high granules threshold
SET
skip_unavailable_shards=1,
allow_experimental_parallel_reading_from_replicas=1,
max_parallel_replicas=3,
use_hedged_requests=0,
cluster_for_parallel_replicas='parallel_replicas',
parallel_replicas_for_non_replicated_merge_tree=1,
parallel_replicas_min_number_of_granules_to_enable=1000;
SELECT '-- count() with parallel replicas -------';
SELECT count() FROM users PREWHERE uid > 2000;
DROP TABLE users;