diff --git a/dbms/programs/copier/CMakeLists.txt b/dbms/programs/copier/CMakeLists.txt index c9f8e44bce8..1be75dff22d 100644 --- a/dbms/programs/copier/CMakeLists.txt +++ b/dbms/programs/copier/CMakeLists.txt @@ -1,5 +1,5 @@ set(CLICKHOUSE_COPIER_SOURCES ${CMAKE_CURRENT_SOURCE_DIR}/ClusterCopier.cpp) -set(CLICKHOUSE_COPIER_LINK PRIVATE clickhouse_functions clickhouse_aggregate_functions daemon) +set(CLICKHOUSE_COPIER_LINK PRIVATE clickhouse_functions clickhouse_table_functions clickhouse_aggregate_functions daemon) #set(CLICKHOUSE_COPIER_INCLUDE SYSTEM PRIVATE ...) clickhouse_program_add(copier) diff --git a/dbms/programs/server/HTTPHandler.cpp b/dbms/programs/server/HTTPHandler.cpp index 6e94d59be07..763a30c1928 100644 --- a/dbms/programs/server/HTTPHandler.cpp +++ b/dbms/programs/server/HTTPHandler.cpp @@ -602,6 +602,8 @@ void HTTPHandler::processQuery( }); } + customizeContext(context); + executeQuery(*in, *used_output.out_maybe_delayed_and_compressed, /* allow_into_outfile = */ false, context, [&response] (const String & content_type) { response.setContentType(content_type); }, [&response] (const String & current_query_id) { response.add("Query-Id", current_query_id); }); diff --git a/dbms/programs/server/HTTPHandler.h b/dbms/programs/server/HTTPHandler.h index 58cb6bb67f1..fb6c9fb532c 100644 --- a/dbms/programs/server/HTTPHandler.h +++ b/dbms/programs/server/HTTPHandler.h @@ -28,6 +28,9 @@ public: void handleRequest(Poco::Net::HTTPServerRequest & request, Poco::Net::HTTPServerResponse & response) override; + /// This method is called right before the query execution. + virtual void customizeContext(DB::Context& /* context */) {} + private: struct Output { diff --git a/dbms/programs/server/TCPHandler.cpp b/dbms/programs/server/TCPHandler.cpp index 013f047ae67..e3582b9ff01 100644 --- a/dbms/programs/server/TCPHandler.cpp +++ b/dbms/programs/server/TCPHandler.cpp @@ -122,7 +122,7 @@ void TCPHandler::runImpl() while (1) { - /// Restore context of request. + /// Set context of request. query_context = connection_context; /// We are waiting for a packet from the client. Thus, every `POLL_INTERVAL` seconds check whether we need to shut down. @@ -158,22 +158,22 @@ void TCPHandler::runImpl() if (!receivePacket()) continue; - query_scope.emplace(query_context); + query_scope.emplace(*query_context); - send_exception_with_stack_trace = query_context.getSettingsRef().calculate_text_stack_trace; + send_exception_with_stack_trace = query_context->getSettingsRef().calculate_text_stack_trace; /// Should we send internal logs to client? if (client_revision >= DBMS_MIN_REVISION_WITH_SERVER_LOGS - && query_context.getSettingsRef().send_logs_level.value != LogsLevel::none) + && query_context->getSettingsRef().send_logs_level.value != LogsLevel::none) { state.logs_queue = std::make_shared(); - state.logs_queue->max_priority = Poco::Logger::parseLevel(query_context.getSettingsRef().send_logs_level.toString()); + state.logs_queue->max_priority = Poco::Logger::parseLevel(query_context->getSettingsRef().send_logs_level.toString()); CurrentThread::attachInternalTextLogsQueue(state.logs_queue); } - query_context.setExternalTablesInitializer([&global_settings, this] (Context & context) + query_context->setExternalTablesInitializer([&global_settings, this] (Context & context) { - if (&context != &query_context) + if (&context != &*query_context) throw Exception("Unexpected context in external tables initializer", ErrorCodes::LOGICAL_ERROR); /// Get blocks of temporary tables @@ -185,9 +185,11 @@ void TCPHandler::runImpl() state.maybe_compressed_in.reset(); /// For more accurate accounting by MemoryTracker. }); + customizeContext(*query_context); + bool may_have_embedded_data = client_revision >= DBMS_MIN_REVISION_WITH_CLIENT_SUPPORT_EMBEDDED_DATA; /// Processing Query - state.io = executeQuery(state.query, query_context, false, state.stage, may_have_embedded_data); + state.io = executeQuery(state.query, *query_context, false, state.stage, may_have_embedded_data); if (state.io.out) state.need_receive_data_for_insert = true; @@ -293,6 +295,9 @@ void TCPHandler::runImpl() LOG_INFO(log, std::fixed << std::setprecision(3) << "Processed in " << watch.elapsedSeconds() << " sec."); + /// It is important to destroy query context here. We do not want it to live arbitrarily longer than the query. + query_context.reset(); + if (network_error) break; } @@ -301,7 +306,7 @@ void TCPHandler::runImpl() void TCPHandler::readData(const Settings & global_settings) { - const auto receive_timeout = query_context.getSettingsRef().receive_timeout.value; + const auto receive_timeout = query_context->getSettingsRef().receive_timeout.value; /// Poll interval should not be greater than receive_timeout const size_t default_poll_interval = global_settings.poll_interval.value * 1000000; @@ -364,8 +369,8 @@ void TCPHandler::processInsertQuery(const Settings & global_settings) /// Send ColumnsDescription for insertion table if (client_revision >= DBMS_MIN_REVISION_WITH_COLUMN_DEFAULTS_METADATA) { - const auto & db_and_table = query_context.getInsertionTable(); - if (auto * columns = ColumnsDescription::loadFromContext(query_context, db_and_table.first, db_and_table.second)) + const auto & db_and_table = query_context->getInsertionTable(); + if (auto * columns = ColumnsDescription::loadFromContext(*query_context, db_and_table.first, db_and_table.second)) sendTableColumns(*columns); } @@ -408,7 +413,7 @@ void TCPHandler::processOrdinaryQuery() } else { - if (after_send_progress.elapsed() / 1000 >= query_context.getSettingsRef().interactive_delay) + if (after_send_progress.elapsed() / 1000 >= query_context->getSettingsRef().interactive_delay) { /// Some time passed and there is a progress. after_send_progress.restart(); @@ -417,7 +422,7 @@ void TCPHandler::processOrdinaryQuery() sendLogs(); - if (async_in.poll(query_context.getSettingsRef().interactive_delay / 1000)) + if (async_in.poll(query_context->getSettingsRef().interactive_delay / 1000)) { /// There is the following result block. block = async_in.read(); @@ -645,11 +650,11 @@ void TCPHandler::receiveQuery() state.is_empty = false; readStringBinary(state.query_id, *in); - query_context.setCurrentQueryId(state.query_id); + query_context->setCurrentQueryId(state.query_id); /// Client info { - ClientInfo & client_info = query_context.getClientInfo(); + ClientInfo & client_info = query_context->getClientInfo(); if (client_revision >= DBMS_MIN_REVISION_WITH_CLIENT_INFO) client_info.read(*in, client_revision); @@ -677,7 +682,7 @@ void TCPHandler::receiveQuery() } /// Per query settings. - Settings & settings = query_context.getSettingsRef(); + Settings & settings = query_context->getSettingsRef(); settings.deserialize(*in); /// Sync timeouts on client and server during current query to avoid dangling queries on server @@ -715,16 +720,16 @@ bool TCPHandler::receiveData() { StoragePtr storage; /// If such a table does not exist, create it. - if (!(storage = query_context.tryGetExternalTable(external_table_name))) + if (!(storage = query_context->tryGetExternalTable(external_table_name))) { NamesAndTypesList columns = block.getNamesAndTypesList(); storage = StorageMemory::create(external_table_name, ColumnsDescription{columns, NamesAndTypesList{}, NamesAndTypesList{}, ColumnDefaults{}, ColumnComments{}, ColumnCodecs{}}); storage->startup(); - query_context.addExternalTable(external_table_name, storage); + query_context->addExternalTable(external_table_name, storage); } /// The data will be written directly to the table. - state.io.out = storage->write(ASTPtr(), query_context); + state.io.out = storage->write(ASTPtr(), *query_context); } if (block) state.io.out->write(block); @@ -763,10 +768,10 @@ void TCPHandler::initBlockOutput(const Block & block) { if (!state.maybe_compressed_out) { - std::string method = query_context.getSettingsRef().network_compression_method; + std::string method = query_context->getSettingsRef().network_compression_method; std::optional level; if (method == "ZSTD") - level = query_context.getSettingsRef().network_zstd_compression_level; + level = query_context->getSettingsRef().network_zstd_compression_level; if (state.compression == Protocol::Compression::Enable) state.maybe_compressed_out = std::make_shared( @@ -802,7 +807,7 @@ bool TCPHandler::isQueryCancelled() if (state.is_cancelled || state.sent_all_data) return true; - if (after_check_cancelled.elapsed() / 1000 < query_context.getSettingsRef().interactive_delay) + if (after_check_cancelled.elapsed() / 1000 < query_context->getSettingsRef().interactive_delay) return false; after_check_cancelled.restart(); diff --git a/dbms/programs/server/TCPHandler.h b/dbms/programs/server/TCPHandler.h index 19641e88d25..38ab27fd67b 100644 --- a/dbms/programs/server/TCPHandler.h +++ b/dbms/programs/server/TCPHandler.h @@ -95,6 +95,9 @@ public: void run(); + /// This method is called right before the query execution. + virtual void customizeContext(DB::Context & /*context*/) {} + private: IServer & server; Poco::Logger * log; @@ -106,7 +109,7 @@ private: UInt64 client_revision = 0; Context connection_context; - Context query_context; + std::optional query_context; /// Streams for reading/writing from/to client connection socket. std::shared_ptr in; diff --git a/dbms/src/Common/ThreadPool.h b/dbms/src/Common/ThreadPool.h index d828de3fea2..a0dae3f810c 100644 --- a/dbms/src/Common/ThreadPool.h +++ b/dbms/src/Common/ThreadPool.h @@ -10,6 +10,7 @@ #include #include +#include #include @@ -133,18 +134,19 @@ public: template explicit ThreadFromGlobalPool(Function && func, Args &&... args) + : state(std::make_shared()) { - mutex = std::make_shared(); - - /// The function object must be copyable, so we wrap lock_guard in shared_ptr. + /// NOTE: If this will throw an exception, the descructor won't be called. GlobalThreadPool::instance().scheduleOrThrow([ - mutex = mutex, - lock = std::make_shared>(*mutex), + state = state, func = std::forward(func), args = std::make_tuple(std::forward(args)...)] { - DB::ThreadStatus thread_status; - std::apply(func, args); + { + DB::ThreadStatus thread_status; + std::apply(func, args); + } + state->set(); }); } @@ -157,7 +159,7 @@ public: { if (joinable()) std::terminate(); - mutex = std::move(rhs.mutex); + state = std::move(rhs.state); return *this; } @@ -171,26 +173,26 @@ public: { if (!joinable()) std::terminate(); - { - std::lock_guard lock(*mutex); - } - mutex.reset(); + + state->wait(); + state.reset(); } void detach() { if (!joinable()) std::terminate(); - mutex.reset(); + state.reset(); } bool joinable() const { - return static_cast(mutex); + return state != nullptr; } private: - std::shared_ptr mutex; /// Object must be moveable. + /// The state used in this object and inside the thread job. + std::shared_ptr state; }; diff --git a/dbms/src/Common/tests/thread_pool_3.cpp b/dbms/src/Common/tests/thread_pool_3.cpp index 687c344288d..924895de308 100644 --- a/dbms/src/Common/tests/thread_pool_3.cpp +++ b/dbms/src/Common/tests/thread_pool_3.cpp @@ -1,4 +1,4 @@ -#include +#include #include #include @@ -10,8 +10,9 @@ void test() { Pool pool(10, 2, 10); + std::mutex mutex; for (size_t i = 0; i < 10; ++i) - pool.schedule([]{ std::cerr << '.'; }); + pool.schedule([&]{ std::lock_guard lock(mutex); std::cerr << '.'; }); pool.wait(); } diff --git a/dbms/src/DataStreams/IBlockInputStream.h b/dbms/src/DataStreams/IBlockInputStream.h index 05b5e059454..8a2b1e9b148 100644 --- a/dbms/src/DataStreams/IBlockInputStream.h +++ b/dbms/src/DataStreams/IBlockInputStream.h @@ -116,7 +116,7 @@ public: */ size_t checkDepth(size_t max_depth) const { return checkDepthImpl(max_depth, max_depth); } - /// Do not allow to change the table while the blocks stream is alive. + /// Do not allow to change the table while the blocks stream and its children are alive. void addTableLock(const TableStructureReadLockPtr & lock) { table_locks.push_back(lock); } /// Get information about execution speed. @@ -242,6 +242,10 @@ public: void enableExtremes() { enabled_extremes = true; } protected: + /// Order is important: `table_locks` must be destroyed after `children` so that tables from + /// which child streams read are protected by the locks during the lifetime of the child streams. + TableStructureReadLocks table_locks; + BlockInputStreams children; std::shared_mutex children_mutex; @@ -268,8 +272,6 @@ protected: } private: - TableStructureReadLocks table_locks; - bool enabled_extremes = false; /// The limit on the number of rows/bytes has been exceeded, and you need to stop execution on the next `read` call, as if the thread has run out. diff --git a/dbms/src/DataTypes/DataTypeArray.cpp b/dbms/src/DataTypes/DataTypeArray.cpp index c1b19d506fe..82de731f4ad 100644 --- a/dbms/src/DataTypes/DataTypeArray.cpp +++ b/dbms/src/DataTypes/DataTypeArray.cpp @@ -498,6 +498,15 @@ bool DataTypeArray::equals(const IDataType & rhs) const } +size_t DataTypeArray::getNumberOfDimensions() const +{ + const DataTypeArray * nested_array = typeid_cast(nested.get()); + if (!nested_array) + return 1; + return 1 + nested_array->getNumberOfDimensions(); /// Every modern C++ compiler optimizes tail recursion. +} + + static DataTypePtr create(const ASTPtr & arguments) { if (!arguments || arguments->children.size() != 1) diff --git a/dbms/src/DataTypes/DataTypeArray.h b/dbms/src/DataTypes/DataTypeArray.h index 58ea41236f9..1451f27dfbe 100644 --- a/dbms/src/DataTypes/DataTypeArray.h +++ b/dbms/src/DataTypes/DataTypeArray.h @@ -112,6 +112,9 @@ public: } const DataTypePtr & getNestedType() const { return nested; } + + /// 1 for plain array, 2 for array of arrays and so on. + size_t getNumberOfDimensions() const; }; } diff --git a/dbms/src/Formats/ODBCDriver2BlockOutputStream.cpp b/dbms/src/Formats/ODBCDriver2BlockOutputStream.cpp index 64d2216dd0e..8e9dbfdd5c2 100644 --- a/dbms/src/Formats/ODBCDriver2BlockOutputStream.cpp +++ b/dbms/src/Formats/ODBCDriver2BlockOutputStream.cpp @@ -3,6 +3,7 @@ #include #include #include +#include namespace DB { @@ -82,8 +83,10 @@ void ODBCDriver2BlockOutputStream::writePrefix() writeODBCString(out, "type"); for (size_t i = 0; i < columns; ++i) { - const ColumnWithTypeAndName & col = header.getByPosition(i); - writeODBCString(out, col.type->getName()); + auto type = header.getByPosition(i).type; + if (type->lowCardinality()) + type = recursiveRemoveLowCardinality(type); + writeODBCString(out, type->getName()); } } diff --git a/dbms/src/Functions/arrayEnumerateDenseRanked.cpp b/dbms/src/Functions/arrayEnumerateDenseRanked.cpp index 05e597e3e73..cc81e35cbe7 100644 --- a/dbms/src/Functions/arrayEnumerateDenseRanked.cpp +++ b/dbms/src/Functions/arrayEnumerateDenseRanked.cpp @@ -4,6 +4,7 @@ namespace DB { + class FunctionArrayEnumerateDenseRanked : public FunctionArrayEnumerateRankedExtended { using Base = FunctionArrayEnumerateRankedExtended; diff --git a/dbms/src/Functions/arrayEnumerateRanked.cpp b/dbms/src/Functions/arrayEnumerateRanked.cpp index ff95344dd57..705d92ecd19 100644 --- a/dbms/src/Functions/arrayEnumerateRanked.cpp +++ b/dbms/src/Functions/arrayEnumerateRanked.cpp @@ -1,61 +1,53 @@ +#include +#include #include "arrayEnumerateRanked.h" + namespace DB { + ArraysDepths getArraysDepths(const ColumnsWithTypeAndName & arguments) { const size_t num_arguments = arguments.size(); + DepthType clear_depth = 1; - DepthType max_array_depth = 0; DepthTypes depths; + /// function signature is the following: + /// f(c0, arr1, c1, arr2, c2, ...) + /// + /// c0 is something called "clear_depth" here. + /// cN... - how deep to look into the corresponding arrN, (called "depths" here) + /// may be omitted - then it means "look at the full depth". + size_t array_num = 0; - DepthType last_array_depth = 0; + DepthType prev_array_depth = 0; for (size_t i = 0; i < num_arguments; ++i) { - const auto type = arguments[i].type; + const DataTypePtr & type = arguments[i].type; + const DataTypeArray * type_array = typeid_cast(type.get()); - if (isArray(type)) + if (type_array) { - if (depths.size() < array_num && last_array_depth) + if (depths.size() < array_num && prev_array_depth) { - depths.emplace_back(last_array_depth); - last_array_depth = 0; + depths.emplace_back(prev_array_depth); + prev_array_depth = 0; } - DepthType depth = 0; - auto sub_type = type; - do - { - auto sub_type_array = typeid_cast(sub_type.get()); - if (!sub_type_array) - break; - sub_type = sub_type_array->getNestedType(); - ++depth; - } while (isArray(sub_type)); - last_array_depth = depth; + prev_array_depth = type_array->getNumberOfDimensions(); ++array_num; } - - if (!arguments[i].column) - continue; - - const IColumn * non_const = nullptr; - if (auto const_array_column = typeid_cast(arguments[i].column.get())) - non_const = const_array_column->getDataColumnPtr().get(); - const auto array = typeid_cast(non_const ? non_const : arguments[i].column.get()); - - if (!array) + else { const auto & depth_column = arguments[i].column; if (depth_column && depth_column->isColumnConst()) { - auto value = depth_column->getUInt(0); + UInt64 value = static_cast(*depth_column).getValue(); if (!value) - throw Exception( - "Arguments for function arrayEnumerateUniqRanked/arrayEnumerateDenseRanked incorrect: depth (" - + std::to_string(value) + ") cant be 0.", + throw Exception("Incorrect arguments for function arrayEnumerateUniqRanked or arrayEnumerateDenseRanked: depth (" + + std::to_string(value) + ") cannot be less or equal 0.", ErrorCodes::BAD_ARGUMENTS); if (i == 0) @@ -65,38 +57,35 @@ ArraysDepths getArraysDepths(const ColumnsWithTypeAndName & arguments) else { if (depths.size() >= array_num) - { - throw Exception( - "Arguments for function arrayEnumerateUniqRanked/arrayEnumerateDenseRanked incorrect: depth (" - + std::to_string(value) + ") for missing array.", + throw Exception("Incorrect arguments for function arrayEnumerateUniqRanked or arrayEnumerateDenseRanked: depth (" + + std::to_string(value) + ") for missing array.", ErrorCodes::BAD_ARGUMENTS); - } + if (value > prev_array_depth) + throw Exception( + "Arguments for function arrayEnumerateUniqRanked/arrayEnumerateDenseRanked incorrect: depth=" + + std::to_string(value) + " for array with depth=" + std::to_string(prev_array_depth) + ".", + ErrorCodes::BAD_ARGUMENTS); + depths.emplace_back(value); } } } } + if (depths.size() < array_num) - { - depths.emplace_back(last_array_depth); - } - - - for (auto & depth : depths) - { - if (max_array_depth < depth) - max_array_depth = depth; - } + depths.emplace_back(prev_array_depth); if (depths.empty()) - throw Exception( - "Arguments for function arrayEnumerateUniqRanked/arrayEnumerateDenseRanked incorrect: At least one array should be passed.", + throw Exception("Incorrect arguments for function arrayEnumerateUniqRanked or arrayEnumerateDenseRanked: at least one array should be passed.", ErrorCodes::BAD_ARGUMENTS); + DepthType max_array_depth = 0; + for (auto depth : depths) + max_array_depth = std::max(depth, max_array_depth); + if (clear_depth > max_array_depth) - throw Exception( - "Arguments for function arrayEnumerateUniqRanked/arrayEnumerateDenseRanked incorrect: clear_depth (" - + std::to_string(clear_depth) + ") cant be larger than max_array_depth (" + std::to_string(max_array_depth) + ").", + throw Exception("Incorrect arguments for function arrayEnumerateUniqRanked or arrayEnumerateDenseRanked: clear_depth (" + + std::to_string(clear_depth) + ") cant be larger than max_array_depth (" + std::to_string(max_array_depth) + ").", ErrorCodes::BAD_ARGUMENTS); return {clear_depth, depths, max_array_depth}; diff --git a/dbms/src/Functions/arrayEnumerateRanked.h b/dbms/src/Functions/arrayEnumerateRanked.h index 8e87d38f025..9056a5cff98 100644 --- a/dbms/src/Functions/arrayEnumerateRanked.h +++ b/dbms/src/Functions/arrayEnumerateRanked.h @@ -12,6 +12,47 @@ #include +/** The function will enumerate distinct values of the passed multidimensional arrays looking inside at the specified depths. + * This is very unusual function made as a special order for Yandex.Metrica. + * + * arrayEnumerateUniqRanked(['hello', 'world', 'hello']) = [1, 1, 2] + * - it returns similar structured array containing number of occurence of the corresponding value. + * + * arrayEnumerateUniqRanked([['hello', 'world'], ['hello'], ['hello']], 1) = [1, 1, 2] + * - look at the depth 1 by default. Elements are ['hello', 'world'], ['hello'], ['hello']. + * + * arrayEnumerateUniqRanked([['hello', 'world'], ['hello'], ['hello']]) = [[1,1],[2],[3]] + * - look at the depth 2. Return similar structured array. + * arrayEnumerateUniqRanked([['hello', 'world'], ['hello'], ['hello']], 2) = [[1,1],[2],[3]] + * - look at the maximum depth by default. + * + * We may pass multiple array arguments. Their elements will be processed as zipped to tuple. + * + * arrayEnumerateUniqRanked(['hello', 'hello', 'world', 'world'], ['a', 'b', 'b', 'b']) = [1, 1, 1, 2] + * + * We may provide arrays of different depths to look at different arguments. + * + * arrayEnumerateUniqRanked([['hello', 'world'], ['hello'], ['world'], ['world']], ['a', 'b', 'b', 'b']) = [[1,1],[1],[1],[2]] + * arrayEnumerateUniqRanked([['hello', 'world'], ['hello'], ['world'], ['world']], 1, ['a', 'b', 'b', 'b'], 1) = [1, 1, 1, 2] + * + * When depths are different, we process less deep arrays as promoted to deeper arrays of similar structure by duplicating elements. + * + * arrayEnumerateUniqRanked( + * [['hello', 'world'], ['hello'], ['world'], ['world']], + * ['a', 'b', 'b', 'b']) + * = arrayEnumerateUniqRanked( + * [['hello', 'world'], ['hello'], ['world'], ['world']], + * [['a', 'a'], ['b'], ['b'], ['b']]) + * + * Finally, we can provide extra first argument named "clear_depth" (it can be considered as 1 by default). + * Array elements at the clear_depth will be enumerated as separate elements (enumeration counter is reset for each new element). + * + * SELECT arrayEnumerateUniqRanked(1, [['hello', 'world'], ['hello'], ['world'], ['world']]) = [[1,1],[2],[2],[3]] + * SELECT arrayEnumerateUniqRanked(2, [['hello', 'world'], ['hello'], ['world'], ['world']]) = [[1,1],[1],[1],[1]] + * SELECT arrayEnumerateUniqRanked(1, [['hello', 'world', 'hello'], ['hello'], ['world'], ['world']]) = [[1,1,2],[3],[2],[3]] + * SELECT arrayEnumerateUniqRanked(2, [['hello', 'world', 'hello'], ['hello'], ['world'], ['world']]) = [[1,1,2],[1],[1],[1]] + */ + namespace DB { namespace ErrorCodes @@ -27,12 +68,21 @@ class FunctionArrayEnumerateDenseRanked; using DepthType = uint32_t; using DepthTypes = std::vector; + struct ArraysDepths { + /// Enumerate elements at the specified level separately. DepthType clear_depth; + + /// Effective depth is the array depth by default or lower value, specified as a constant argument following the array. + /// f([[1, 2], [3]]) - effective depth is 2. + /// f([[1, 2], [3]], 1) - effective depth is 1. DepthTypes depths; + + /// Maximum effective depth. DepthType max_array_depth; }; + /// Return depth info about passed arrays ArraysDepths getArraysDepths(const ColumnsWithTypeAndName & arguments); @@ -55,7 +105,9 @@ public: + ", should be at least 1.", ErrorCodes::NUMBER_OF_ARGUMENTS_DOESNT_MATCH); - const auto & arrays_depths = getArraysDepths(arguments); + const ArraysDepths arrays_depths = getArraysDepths(arguments); + + /// Return type is the array of the depth as the maximum effective depth of arguments, containing UInt32. DataTypePtr type = std::make_shared(); for (DepthType i = 0; i < arrays_depths.max_array_depth; ++i) @@ -79,15 +131,15 @@ private: /// Hash a set of keys into a UInt128 value. -static inline UInt128 ALWAYS_INLINE hash128depths(const std::vector & indexes, const ColumnRawPtrs & key_columns) +static inline UInt128 ALWAYS_INLINE hash128depths(const std::vector & indices, const ColumnRawPtrs & key_columns) { UInt128 key; SipHash hash; for (size_t j = 0, keys_size = key_columns.size(); j < keys_size; ++j) { - // Debug: const auto & field = (*key_columns[j])[indexes[j]]; DUMP(j, indexes[j], field); - key_columns[j]->updateHashWithValue(indexes[j], hash); + // Debug: const auto & field = (*key_columns[j])[indices[j]]; DUMP(j, indices[j], field); + key_columns[j]->updateHashWithValue(indices[j], hash); } hash.get128(key.low, key.high); @@ -111,9 +163,11 @@ void FunctionArrayEnumerateRankedExtended::executeImpl( for (size_t i = 0; i < arguments.size(); ++i) args.emplace_back(block.getByPosition(arguments[i])); - const auto & arrays_depths = getArraysDepths(args); + const ArraysDepths arrays_depths = getArraysDepths(args); - auto get_array_column = [&](const auto & column) -> const DB::ColumnArray * { + /// If the column is Array - return it. If the const Array - materialize it, keep ownership and return. + auto get_array_column = [&](const auto & column) -> const DB::ColumnArray * + { const ColumnArray * array = checkAndGetColumn(column); if (!array) { @@ -146,7 +200,7 @@ void FunctionArrayEnumerateRankedExtended::executeImpl( if (*offsets_by_depth[0] != array->getOffsets()) { throw Exception( - "Lengths and depths of all arrays passed to " + getName() + " must be equal.", + "Lengths and effective depths of all arrays passed to " + getName() + " must be equal.", ErrorCodes::SIZES_OF_ARRAYS_DOESNT_MATCH); } } @@ -170,7 +224,7 @@ void FunctionArrayEnumerateRankedExtended::executeImpl( if (*offsets_by_depth[col_depth] != array->getOffsets()) { throw Exception( - "Lengths and depths of all arrays passed to " + getName() + " must be equal.", + "Lengths and effective depths of all arrays passed to " + getName() + " must be equal.", ErrorCodes::SIZES_OF_ARRAYS_DOESNT_MATCH); } } @@ -180,7 +234,7 @@ void FunctionArrayEnumerateRankedExtended::executeImpl( { throw Exception( getName() + ": Passed array number " + std::to_string(array_num) + " depth (" - + std::to_string(arrays_depths.depths[array_num]) + ") more than actual array depth (" + std::to_string(col_depth) + + std::to_string(arrays_depths.depths[array_num]) + ") is more than the actual array depth (" + std::to_string(col_depth) + ").", ErrorCodes::SIZES_OF_ARRAYS_DOESNT_MATCH); } @@ -251,6 +305,7 @@ void FunctionArrayEnumerateRankedExtended::executeMethodImpl( const ArraysDepths & arrays_depths, ColumnUInt32::Container & res_values) { + /// Offsets at the depth we want to look. const size_t current_offset_depth = arrays_depths.max_array_depth; const auto & offsets = *offsets_by_depth[current_offset_depth - 1]; @@ -264,22 +319,24 @@ void FunctionArrayEnumerateRankedExtended::executeMethodImpl( HashTableAllocatorWithStackMemory<(1ULL << INITIAL_SIZE_DEGREE) * sizeof(UInt128)>>; Map indices; - std::vector indexes_by_depth(arrays_depths.max_array_depth); + std::vector indices_by_depth(arrays_depths.max_array_depth); std::vector current_offset_n_by_depth(arrays_depths.max_array_depth); UInt32 rank = 0; - std::vector columns_indexes(columns.size()); + std::vector columns_indices(columns.size()); + for (size_t off : offsets) { bool want_clear = false; + /// For each element at the depth we want to look. for (size_t j = prev_off; j < off; ++j) { for (size_t col_n = 0; col_n < columns.size(); ++col_n) - columns_indexes[col_n] = indexes_by_depth[arrays_depths.depths[col_n] - 1]; + columns_indices[col_n] = indices_by_depth[arrays_depths.depths[col_n] - 1]; - auto hash = hash128depths(columns_indexes, columns); + auto hash = hash128depths(columns_indices, columns); if constexpr (std::is_same_v) { @@ -297,13 +354,13 @@ void FunctionArrayEnumerateRankedExtended::executeMethodImpl( res_values[j] = idx; } - // Debug: DUMP(off, prev_off, j, columns_indexes, res_values[j], columns); + // Debug: DUMP(off, prev_off, j, columns_indices, res_values[j], columns); for (int depth = current_offset_depth - 1; depth >= 0; --depth) { - ++indexes_by_depth[depth]; + ++indices_by_depth[depth]; - if (indexes_by_depth[depth] == (*offsets_by_depth[depth])[current_offset_n_by_depth[depth]]) + if (indices_by_depth[depth] == (*offsets_by_depth[depth])[current_offset_n_by_depth[depth]]) { if (static_cast(arrays_depths.clear_depth) == depth + 1) want_clear = true; @@ -315,6 +372,7 @@ void FunctionArrayEnumerateRankedExtended::executeMethodImpl( } } } + if (want_clear) { want_clear = false; diff --git a/dbms/src/Interpreters/Context.cpp b/dbms/src/Interpreters/Context.cpp index 8c2e2b8178c..b7613116f2d 100644 --- a/dbms/src/Interpreters/Context.cpp +++ b/dbms/src/Interpreters/Context.cpp @@ -1818,6 +1818,19 @@ void Context::addXDBCBridgeCommand(std::unique_ptr cmd) shared->bridge_commands.emplace_back(std::move(cmd)); } + +IHostContextPtr & Context::getHostContext() +{ + return host_context; +} + + +const IHostContextPtr & Context::getHostContext() const +{ + return host_context; +} + + std::shared_ptr Context::getActionLocksManager() { auto lock = getLock(); diff --git a/dbms/src/Interpreters/Context.h b/dbms/src/Interpreters/Context.h index df1e23d885b..8b4e66094ff 100644 --- a/dbms/src/Interpreters/Context.h +++ b/dbms/src/Interpreters/Context.h @@ -99,6 +99,15 @@ using TableAndCreateASTs = std::map; /// Callback for external tables initializer using ExternalTablesInitializer = std::function; +/// An empty interface for an arbitrary object that may be attached by a shared pointer +/// to query context, when using ClickHouse as a library. +struct IHostContext +{ + virtual ~IHostContext() = default; +}; + +using IHostContextPtr = std::shared_ptr; + /** A set of known objects that can be used in the query. * Consists of a shared part (always common to all sessions and queries) * and copied part (which can be its own for each session or query). @@ -139,6 +148,12 @@ private: using DatabasePtr = std::shared_ptr; using Databases = std::map>; + IHostContextPtr host_context; /// Arbitrary object that may used to attach some host specific information to query context, + /// when using ClickHouse as a library in some project. For example, it may contain host + /// logger, some query identification information, profiling guards, etc. This field is + /// to be customized in HTTP and TCP servers by overloading the customizeContext(DB::Context&) + /// methods. + /// Use copy constructor or createGlobal() instead Context(); @@ -452,6 +467,9 @@ public: /// Add started bridge command. It will be killed after context destruction void addXDBCBridgeCommand(std::unique_ptr cmd); + IHostContextPtr & getHostContext(); + const IHostContextPtr & getHostContext() const; + private: /** Check if the current client has access to the specified database. * If access is denied, throw an exception. diff --git a/dbms/src/Interpreters/Settings.h b/dbms/src/Interpreters/Settings.h index 16bceb90d76..20e07d15916 100644 --- a/dbms/src/Interpreters/Settings.h +++ b/dbms/src/Interpreters/Settings.h @@ -103,7 +103,7 @@ struct Settings \ M(SettingUInt64, optimize_min_equality_disjunction_chain_length, 3, "The minimum length of the expression `expr = x1 OR ... expr = xN` for optimization ") \ \ - M(SettingUInt64, min_bytes_to_use_direct_io, 0, "The minimum number of bytes for input/output operations is bypassing the page cache. 0 - disabled.") \ + M(SettingUInt64, min_bytes_to_use_direct_io, 0, "The minimum number of bytes for reading the data with O_DIRECT option during SELECT queries execution. 0 - disabled.") \ \ M(SettingBool, force_index_by_date, 0, "Throw an exception if there is a partition key in a table, and it is not used.") \ M(SettingBool, force_primary_key, 0, "Throw an exception if there is primary key in a table, and it is not used.") \ diff --git a/dbms/src/Storages/MergeTree/MergeTreeIndices.h b/dbms/src/Storages/MergeTree/MergeTreeIndices.h index 6738d667b44..1d62e9e9e9c 100644 --- a/dbms/src/Storages/MergeTree/MergeTreeIndices.h +++ b/dbms/src/Storages/MergeTree/MergeTreeIndices.h @@ -11,6 +11,7 @@ #include #include #include +#include constexpr auto INDEX_FILE_PREFIX = "skp_idx_"; diff --git a/dbms/src/Storages/MergeTree/MergeTreeSetSkippingIndex.cpp b/dbms/src/Storages/MergeTree/MergeTreeSetSkippingIndex.cpp index 69323fe8bb4..227c6bfd7f7 100644 --- a/dbms/src/Storages/MergeTree/MergeTreeSetSkippingIndex.cpp +++ b/dbms/src/Storages/MergeTree/MergeTreeSetSkippingIndex.cpp @@ -18,7 +18,7 @@ namespace ErrorCodes } /// 0b11 -- can be true and false at the same time -const Field UNKNOWN_FIELD(3); +const Field UNKNOWN_FIELD(3u); MergeTreeSetIndexGranule::MergeTreeSetIndexGranule(const MergeTreeSetSkippingIndex & index) @@ -47,7 +47,16 @@ void MergeTreeSetIndexGranule::serializeBinary(WriteBuffer & ostr) const for (size_t i = 0; i < index.columns.size(); ++i) { const auto & type = index.data_types[i]; - type->serializeBinaryBulk(*columns[i], ostr, 0, size()); + + IDataType::SerializeBinaryBulkSettings settings; + settings.getter = [&ostr](IDataType::SubstreamPath) -> WriteBuffer * { return &ostr; }; + settings.position_independent_encoding = false; + settings.low_cardinality_max_dictionary_size = 0; + + IDataType::SerializeBinaryBulkStatePtr state; + type->serializeBinaryBulkStatePrefix(settings, state); + type->serializeBinaryBulkWithMultipleStreams(*columns[i], 0, size(), settings, state); + type->serializeBinaryBulkStateSuffix(settings, state); } } @@ -66,11 +75,21 @@ void MergeTreeSetIndexGranule::deserializeBinary(ReadBuffer & istr) size_type->deserializeBinary(field_rows, istr); size_t rows_to_read = field_rows.get(); + if (rows_to_read == 0) + return; + for (size_t i = 0; i < index.columns.size(); ++i) { const auto & type = index.data_types[i]; auto new_column = type->createColumn(); - type->deserializeBinaryBulk(*new_column, istr, rows_to_read, 0); + + IDataType::DeserializeBinaryBulkSettings settings; + settings.getter = [&](IDataType::SubstreamPath) -> ReadBuffer * { return &istr; }; + settings.position_independent_encoding = false; + + IDataType::DeserializeBinaryBulkStatePtr state; + type->deserializeBinaryBulkStatePrefix(settings, state); + type->deserializeBinaryBulkWithMultipleStreams(*new_column, rows_to_read, settings, state); block.insert(ColumnWithTypeAndName(new_column->getPtr(), type, index.columns[i])); } @@ -177,10 +196,24 @@ bool SetIndexCondition::mayBeTrueOnGranule(MergeTreeIndexGranulePtr idx_granule) Block result = granule->getElementsBlock(); actions->execute(result); - const auto & column = result.getByName(expression_ast->getColumnName()).column; + auto column = result.getByName(expression_ast->getColumnName()).column->convertToFullColumnIfLowCardinality(); + auto * col_uint8 = typeid_cast(column.get()); + + const NullMap * null_map = nullptr; + + if (auto * col_nullable = typeid_cast(column.get())) + { + col_uint8 = typeid_cast(&col_nullable->getNestedColumn()); + null_map = &col_nullable->getNullMapData(); + } + + if (!col_uint8) + throw Exception("ColumnUInt8 expected as Set index condition result.", ErrorCodes::LOGICAL_ERROR); + + auto & condition = col_uint8->getData(); for (size_t i = 0; i < column->size(); ++i) - if (column->getInt(i) & 1) + if ((!null_map || (*null_map)[i] == 0) && condition[i] & 1) return true; return false; diff --git a/dbms/tests/queries/0_stateless/00907_set_index_with_nullable_and_low_cardinality.reference b/dbms/tests/queries/0_stateless/00907_set_index_with_nullable_and_low_cardinality.reference new file mode 100644 index 00000000000..f3918a338c4 --- /dev/null +++ b/dbms/tests/queries/0_stateless/00907_set_index_with_nullable_and_low_cardinality.reference @@ -0,0 +1,30 @@ +1 a +- +2 b +- +-- +1 a +- +2 b +- +-- +1 a +- +2 b +- +---- +1 a +- +2 b +- +-- +1 a +- +2 b +- +-- +1 a +- +2 b +- +---- diff --git a/dbms/tests/queries/0_stateless/00907_set_index_with_nullable_and_low_cardinality.sql b/dbms/tests/queries/0_stateless/00907_set_index_with_nullable_and_low_cardinality.sql new file mode 100644 index 00000000000..363e0e9c0b5 --- /dev/null +++ b/dbms/tests/queries/0_stateless/00907_set_index_with_nullable_and_low_cardinality.sql @@ -0,0 +1,69 @@ +SET allow_experimental_data_skipping_indices=1; + +drop table if exists test.nullable_set_index; +create table test.nullable_set_index (a UInt64, b Nullable(String), INDEX b_index b TYPE set(0) GRANULARITY 8192) engine = MergeTree order by a; +insert into test.nullable_set_index values (1, 'a'); +insert into test.nullable_set_index values (2, 'b'); +select * from test.nullable_set_index where b = 'a'; +select '-'; +select * from test.nullable_set_index where b = 'b'; +select '-'; +select * from test.nullable_set_index where b = 'c'; +select '--'; + +drop table if exists test.nullable_set_index; +create table test.nullable_set_index (a UInt64, b Nullable(String), INDEX b_index b TYPE set(1) GRANULARITY 8192) engine = MergeTree order by a; +insert into test.nullable_set_index values (1, 'a'); +insert into test.nullable_set_index values (2, 'b'); +select * from test.nullable_set_index where b = 'a'; +select '-'; +select * from test.nullable_set_index where b = 'b'; +select '-'; +select * from test.nullable_set_index where b = 'c'; +select '--'; + +drop table if exists test.nullable_set_index; +create table test.nullable_set_index (a UInt64, b Nullable(String), INDEX b_index b TYPE set(0) GRANULARITY 8192) engine = MergeTree order by a; +insert into test.nullable_set_index values (1, 'a'), (2, 'b'); +select * from test.nullable_set_index where b = 'a'; +select '-'; +select * from test.nullable_set_index where b = 'b'; +select '-'; +select * from test.nullable_set_index where b = 'c'; +select '----'; + + +drop table if exists test.nullable_set_index; +create table test.nullable_set_index (a UInt64, b LowCardinality(Nullable(String)), INDEX b_index b TYPE set(0) GRANULARITY 8192) engine = MergeTree order by a; +insert into test.nullable_set_index values (1, 'a'); +insert into test.nullable_set_index values (2, 'b'); +select * from test.nullable_set_index where b = 'a'; +select '-'; +select * from test.nullable_set_index where b = 'b'; +select '-'; +select * from test.nullable_set_index where b = 'c'; +select '--'; + +drop table if exists test.nullable_set_index; +create table test.nullable_set_index (a UInt64, b LowCardinality(Nullable(String)), INDEX b_index b TYPE set(1) GRANULARITY 8192) engine = MergeTree order by a; +insert into test.nullable_set_index values (1, 'a'); +insert into test.nullable_set_index values (2, 'b'); +select * from test.nullable_set_index where b = 'a'; +select '-'; +select * from test.nullable_set_index where b = 'b'; +select '-'; +select * from test.nullable_set_index where b = 'c'; +select '--'; + +drop table if exists test.nullable_set_index; +create table test.nullable_set_index (a UInt64, b LowCardinality(Nullable(String)), INDEX b_index b TYPE set(0) GRANULARITY 8192) engine = MergeTree order by a; +insert into test.nullable_set_index values (1, 'a'), (2, 'b'); +select * from test.nullable_set_index where b = 'a'; +select '-'; +select * from test.nullable_set_index where b = 'b'; +select '-'; +select * from test.nullable_set_index where b = 'c'; +select '----'; + +drop table if exists test.nullable_set_index; + diff --git a/dbms/tests/queries/0_stateless/00907_set_index_with_nullable_and_low_cardinality_bug.reference b/dbms/tests/queries/0_stateless/00907_set_index_with_nullable_and_low_cardinality_bug.reference new file mode 100644 index 00000000000..051c3fcdbcf --- /dev/null +++ b/dbms/tests/queries/0_stateless/00907_set_index_with_nullable_and_low_cardinality_bug.reference @@ -0,0 +1 @@ +subscribe alice diff --git a/dbms/tests/queries/0_stateless/00907_set_index_with_nullable_and_low_cardinality_bug.sql b/dbms/tests/queries/0_stateless/00907_set_index_with_nullable_and_low_cardinality_bug.sql new file mode 100644 index 00000000000..5bec1210a1e --- /dev/null +++ b/dbms/tests/queries/0_stateless/00907_set_index_with_nullable_and_low_cardinality_bug.sql @@ -0,0 +1,20 @@ +SET allow_experimental_data_skipping_indices=1; + +drop table if exists test.null_lc_set_index; + +CREATE TABLE test.null_lc_set_index ( + timestamp DateTime, + action LowCardinality(Nullable(String)), + user LowCardinality(Nullable(String)), + INDEX test_user_idx (user) TYPE set(0) GRANULARITY 8192 +) ENGINE=MergeTree + PARTITION BY toYYYYMMDD(timestamp) + ORDER BY (timestamp, action, cityHash64(user)) + SAMPLE BY cityHash64(user); +INSERT INTO test.null_lc_set_index VALUES (1550883010, 'subscribe', 'alice'); +INSERT INTO test.null_lc_set_index VALUES (1550883020, 'follow', 'bob'); + +SELECT action, user FROM test.null_lc_set_index WHERE user = 'alice'; + +drop table if exists test.null_lc_set_index; + diff --git a/dbms/tests/queries/0_stateless/00909_arrayEnumerateUniq.sql b/dbms/tests/queries/0_stateless/00909_arrayEnumerateUniq.sql index 0682da084c1..d622b1acd76 100644 --- a/dbms/tests/queries/0_stateless/00909_arrayEnumerateUniq.sql +++ b/dbms/tests/queries/0_stateless/00909_arrayEnumerateUniq.sql @@ -145,11 +145,11 @@ SELECT arrayEnumerateUniqRanked(); -- { serverError 42 } SELECT arrayEnumerateUniqRanked([]); SELECT arrayEnumerateUniqRanked(1); -- { serverError 36 } SELECT arrayEnumerateUniqRanked(2,[]); -- { serverError 36 } -SELECT arrayEnumerateUniqRanked(2,[],2); -- { serverError 190 } +SELECT arrayEnumerateUniqRanked(2,[],2); -- { serverError 36 } SELECT arrayEnumerateUniqRanked(2,[],[]); -- { serverError 36 } -SELECT arrayEnumerateUniqRanked(2,[],[],3); -- { serverError 190 } -SELECT arrayEnumerateUniqRanked([],2); -- { serverError 190 } -SELECT arrayEnumerateUniqRanked([],2,[]); -- { serverError 190 } +SELECT arrayEnumerateUniqRanked(2,[],[],3); -- { serverError 36 } +SELECT arrayEnumerateUniqRanked([],2); -- { serverError 36 } +SELECT arrayEnumerateUniqRanked([],2,[]); -- { serverError 36 } SELECT arrayEnumerateUniqRanked(0,[],0); -- { serverError 36 } SELECT arrayEnumerateUniqRanked(0,0,0); -- { serverError 36 } SELECT arrayEnumerateUniqRanked(1,1,1); -- { serverError 36 } @@ -170,5 +170,13 @@ SELECT arrayEnumerateUniqRanked([1,2], 1, 2); -- { serverError 36 } SELECT arrayEnumerateUniqRanked([1,2], 1, 3, 4, 5); -- { serverError 36 } SELECT arrayEnumerateUniqRanked([1,2], 1, 3, [4], 5); -- { serverError 36 } SELECT arrayEnumerateDenseRanked([[[[[[[[[[42]]]]]]]]]]); -SELECT arrayEnumerateUniqRanked('wat', [1,2]); -- { serverError 48 } -SELECT arrayEnumerateUniqRanked(1, [1,2], 'boom'); -- { serverError 48 } +SELECT arrayEnumerateUniqRanked('wat', [1,2]); -- { serverError 170 } +SELECT arrayEnumerateUniqRanked(1, [1,2], 'boom'); -- { serverError 170 } +SELECT arrayEnumerateDenseRanked(['\0'], -8363126); -- { serverError 170 } +SELECT arrayEnumerateDenseRanked(-10, ['\0'], -8363126); -- { serverError 170 } +SELECT arrayEnumerateDenseRanked(1, ['\0'], -8363126); -- { serverError 170 } +SELECT arrayEnumerateDenseRanked(-101, ['\0']); -- { serverError 170 } +SELECT arrayEnumerateDenseRanked(1.1, [10,20,10,30]); -- { serverError 170 } +SELECT arrayEnumerateDenseRanked([10,20,10,30], 0.4); -- { serverError 170 } +SELECT arrayEnumerateDenseRanked([10,20,10,30], 1.8); -- { serverError 170 } +SELECT arrayEnumerateUniqRanked(1, [], 1000000000); -- { serverError 36 } diff --git a/dbms/tests/queries/0_stateless/00909_kill_not_initialized_query.reference b/dbms/tests/queries/0_stateless/00909_kill_not_initialized_query.reference index 573541ac970..aa47d0d46d4 100644 --- a/dbms/tests/queries/0_stateless/00909_kill_not_initialized_query.reference +++ b/dbms/tests/queries/0_stateless/00909_kill_not_initialized_query.reference @@ -1 +1,2 @@ 0 +0 diff --git a/dbms/tests/queries/0_stateless/00909_kill_not_initialized_query.sh b/dbms/tests/queries/0_stateless/00909_kill_not_initialized_query.sh index 443b2463a33..76d6424e690 100755 --- a/dbms/tests/queries/0_stateless/00909_kill_not_initialized_query.sh +++ b/dbms/tests/queries/0_stateless/00909_kill_not_initialized_query.sh @@ -10,20 +10,37 @@ $CLICKHOUSE_CLIENT -q "DROP TABLE IF EXISTS test.cannot_kill_query" $CLICKHOUSE_CLIENT -q "CREATE TABLE test.cannot_kill_query (x UInt64) ENGINE = MergeTree ORDER BY x" &> /dev/null $CLICKHOUSE_CLIENT -q "INSERT INTO test.cannot_kill_query SELECT * FROM numbers(10000000)" &> /dev/null +# This SELECT query will run for a long time. It's used as bloker for ALTER query. It will be killed with SYNC kill. query_for_pending="SELECT count() FROM test.cannot_kill_query WHERE NOT ignore(sleep(1)) SETTINGS max_threads=1, max_block_size=1" $CLICKHOUSE_CLIENT -q "$query_for_pending" &>/dev/null & +sleep 1 # queries should be in strict order + +# This ALTER query will wait until $query_for_pending finished. Also it will block $query_to_kill. $CLICKHOUSE_CLIENT -q "ALTER TABLE test.cannot_kill_query MODIFY COLUMN x UInt64" &>/dev/null & +sleep 1 + +# This SELECT query will also run for a long time. Also it's blocked by ALTER query. It will be killed with ASYNC kill. +# This is main idea which we check -- blocked queries can be killed with ASYNC kill. query_to_kill="SELECT sum(1) FROM test.cannot_kill_query WHERE NOT ignore(sleep(1)) SETTINGS max_threads=1" $CLICKHOUSE_CLIENT -q "$query_to_kill" &>/dev/null & -sleep 3 # just to be sure that 'KILL ...' will be executed after 'SELECT ... WHERE NOT ignore(sleep(1))' +sleep 1 # just to be sure that kill of $query_to_kill will be executed after $query_to_kill. -timeout 15 $CLICKHOUSE_CLIENT -q "KILL QUERY WHERE query='$query_to_kill' SYNC" &>/dev/null +# Kill $query_to_kill with ASYNC kill. We will check that information about KILL is not lost. +$CLICKHOUSE_CLIENT -q "KILL QUERY WHERE query='$query_to_kill' ASYNC" &>/dev/null +sleep 1 + +# Kill $query_for_pending SYNC. This query is not blocker, so it should be killed fast. +timeout 5 $CLICKHOUSE_CLIENT -q "KILL QUERY WHERE query='$query_for_pending' SYNC" &>/dev/null + +# But let's sleep a little time, just to be sure +sleep 3 + +# Both queries have to be killed, doesn't matter with SYNC or ASYNC kill +$CLICKHOUSE_CLIENT -q "SELECT count() FROM system.processes where query='$query_for_pending'" $CLICKHOUSE_CLIENT -q "SELECT count() FROM system.processes where query='$query_to_kill'" -$CLICKHOUSE_CLIENT -q "KILL QUERY WHERE query='$query_for_pending'" &>/dev/null & # kill pending query - $CLICKHOUSE_CLIENT -q "DROP TABLE IF EXISTS test.cannot_kill_query" &>/dev/null diff --git a/debian/changelog.in b/debian/changelog.in index 4a74214c022..ce2ee757999 100644 --- a/debian/changelog.in +++ b/debian/changelog.in @@ -1,4 +1,4 @@ -clickhouse (2:@VERSION_STRING@) unstable; urgency=low +clickhouse (@VERSION_STRING@) unstable; urgency=low * Modified source code diff --git a/docs/en/interfaces/third-party/client_libraries.md b/docs/en/interfaces/third-party/client_libraries.md index 5878d6e0200..a5b9991a418 100644 --- a/docs/en/interfaces/third-party/client_libraries.md +++ b/docs/en/interfaces/third-party/client_libraries.md @@ -32,6 +32,7 @@ - [RClickhouse](https://github.com/IMSMWU/RClickhouse) - Java - [clickhouse-client-java](https://github.com/VirtusAI/clickhouse-client-java) + - [clickhouse-client](https://github.com/Ecwid/clickhouse-client) - Scala - [clickhouse-scala-client](https://github.com/crobox/clickhouse-scala-client) - Kotlin diff --git a/docs/en/operations/backup.md b/docs/en/operations/backup.md index 458817e27d7..f6cb381e602 100644 --- a/docs/en/operations/backup.md +++ b/docs/en/operations/backup.md @@ -31,5 +31,4 @@ For more information about queries related to partition manipulations, see the [ A third-party tool is available to automate this approach: [clickhouse-backup](https://github.com/AlexAkulov/clickhouse-backup). - [Original article](https://clickhouse.yandex/docs/en/operations/backup/) diff --git a/docs/en/operations/server_settings/settings.md b/docs/en/operations/server_settings/settings.md index f339fb6ce28..d853e0e65e8 100644 --- a/docs/en/operations/server_settings/settings.md +++ b/docs/en/operations/server_settings/settings.md @@ -196,7 +196,7 @@ For more details, see [GraphiteMergeTree](../../operations/table_engines/graphit The port for connecting to the server over HTTP(s). -If `https_port` is specified, [openSSL](#openssl) must be configured. +If `https_port` is specified, [openSSL](#server_settings-openssl) must be configured. If `http_port` is specified, the openSSL configuration is ignored even if it is set. @@ -417,7 +417,7 @@ The value 0 means that you can delete all tables without any restrictions. ## merge_tree {#server_settings-merge_tree} -Fine tuning for tables in the [ MergeTree](../../operations/table_engines/mergetree.md). +Fine tuning for tables in the [MergeTree](../../operations/table_engines/mergetree.md). For more information, see the MergeTreeSettings.h header file. @@ -430,7 +430,7 @@ For more information, see the MergeTreeSettings.h header file. ``` -## openSSL +## openSSL {#server_settings-openssl} SSL client/server configuration. @@ -609,6 +609,19 @@ Port for communicating with clients over the TCP protocol. 9000 ``` +## tcp_port_secure {#server_settings-tcp_port_secure} + +Port for communicating with the clients over the secure connection by TCP protocol. Use it with [OpenSSL](#server_settings-openssl) settings. + +**Possible values** + +Positive integer. + +**Default value** + +```xml +9440 +``` ## tmp_path diff --git a/docs/en/operations/settings/settings.md b/docs/en/operations/settings/settings.md index e30758fc943..fff660999e2 100644 --- a/docs/en/operations/settings/settings.md +++ b/docs/en/operations/settings/settings.md @@ -175,6 +175,20 @@ Any positive integer. **Default value**: 1048576. +## min_bytes_to_use_direct_io {#settings-min_bytes_to_use_direct_io} + +The minimum data volume to be read from storage required for using of the direct I/O access to the storage disk. + +ClickHouse uses this setting when selecting the data from tables. If summary storage volume of all the data to be read exceeds `min_bytes_to_use_direct_io` bytes, then ClickHouse reads the data from the storage disk with `O_DIRECT` option. + +**Possible values** + +Positive integer. + +0 — The direct I/O is disabled. + +**Default value**: 0. + ## log_queries Setting up query logging. diff --git a/docs/en/operations/table_engines/kafka.md b/docs/en/operations/table_engines/kafka.md index 3927e472e50..87322afb48e 100644 --- a/docs/en/operations/table_engines/kafka.md +++ b/docs/en/operations/table_engines/kafka.md @@ -9,38 +9,38 @@ Kafka lets you: - Process streams as they become available. -Old format: +## Creating a Table {#table_engine-kafka-creating-a-table} ``` -Kafka(kafka_broker_list, kafka_topic_list, kafka_group_name, kafka_format - [, kafka_row_delimiter, kafka_schema, kafka_num_consumers]) +CREATE TABLE [IF NOT EXISTS] [db.]table_name [ON CLUSTER cluster] +( + name1 [type1] [DEFAULT|MATERIALIZED|ALIAS expr1], + name2 [type2] [DEFAULT|MATERIALIZED|ALIAS expr2], + ... +) ENGINE = Kafka() +SETTINGS + kafka_broker_list = 'host:port', + kafka_topic_list = 'topic1,topic2,...', + kafka_group_name = 'group_name', + kafka_format = 'data_format'[,] + [kafka_row_delimiter = 'delimiter_symbol',] + [kafka_schema = '',] + [kafka_num_consumers = N,] + [kafka_skip_broken_messages = <0|1>] ``` - -New format: - -``` -Kafka SETTINGS - kafka_broker_list = 'localhost:9092', - kafka_topic_list = 'topic1,topic2', - kafka_group_name = 'group1', - kafka_format = 'JSONEachRow', - kafka_row_delimiter = '\n', - kafka_schema = '', - kafka_num_consumers = 2 -``` - Required parameters: -- `kafka_broker_list` – A comma-separated list of brokers (`localhost:9092`). -- `kafka_topic_list` – A list of Kafka topics (`my_topic`). -- `kafka_group_name` – A group of Kafka consumers (`group1`). Reading margins are tracked for each group separately. If you don't want messages to be duplicated in the cluster, use the same group name everywhere. -- `kafka_format` – Message format. Uses the same notation as the SQL ` FORMAT` function, such as ` JSONEachRow`. For more information, see the "Formats" section. +- `kafka_broker_list` – A comma-separated list of brokers (for example, `localhost:9092`). +- `kafka_topic_list` – A list of Kafka topics. +- `kafka_group_name` – A group of Kafka consumers. Reading margins are tracked for each group separately. If you don't want messages to be duplicated in the cluster, use the same group name everywhere. +- `kafka_format` – Message format. Uses the same notation as the SQL `FORMAT` function, such as ` JSONEachRow`. For more information, see the [Formats](../../interfaces/formats.md) section. Optional parameters: -- `kafka_row_delimiter` - Character-delimiter of records (rows), which ends the message. -- `kafka_schema` – An optional parameter that must be used if the format requires a schema definition. For example, [Cap'n Proto](https://capnproto.org/) requires the path to the schema file and the name of the root `schema.capnp:Message` object. +- `kafka_row_delimiter` – Delimiter character, which ends the message. +- `kafka_schema` – Parameter that must be used if the format requires a schema definition. For example, [Cap'n Proto](https://capnproto.org/) requires the path to the schema file and the name of the root `schema.capnp:Message` object. - `kafka_num_consumers` – The number of consumers per table. Default: `1`. Specify more consumers if the throughput of one consumer is insufficient. The total number of consumers should not exceed the number of partitions in the topic, since only one consumer can be assigned per partition. +- `kafka_skip_broken_messages` – Mode of Kafka messages parser. If `kafka_skip_broken_messages = 1` then the engine skips the Kafka messages (message equals a row of data) that can't be parsed. Examples: @@ -72,6 +72,23 @@ Examples: kafka_num_consumers = 4; ``` + +
Deprecated Method for Creating a Table + +!!! attention + Do not use this method in new projects and, if possible, switch the old projects to the method described above. + + +``` +Kafka(kafka_broker_list, kafka_topic_list, kafka_group_name, kafka_format + [, kafka_row_delimiter, kafka_schema, kafka_num_consumers, kafka_skip_broken_messages]) +``` + +
+ +## Description + + The delivered messages are tracked automatically, so each message in a group is only counted once. If you want to get the data twice, then create a copy of the table with another group name. Groups are flexible and synced on the cluster. For instance, if you have 10 topics and 5 copies of a table in a cluster, then each copy gets 2 topics. If the number of copies changes, the topics are redistributed across the copies automatically. Read more about this at [http://kafka.apache.org/intro](http://kafka.apache.org/intro). diff --git a/docs/en/operations/table_engines/mergetree.md b/docs/en/operations/table_engines/mergetree.md index 7d86776d27b..53bb909b16d 100644 --- a/docs/en/operations/table_engines/mergetree.md +++ b/docs/en/operations/table_engines/mergetree.md @@ -70,6 +70,8 @@ For a description of request parameters, see [request description](../../query_l - `SETTINGS` — Additional parameters that control the behavior of the `MergeTree`: - `index_granularity` — The granularity of an index. The number of data rows between the "marks" of an index. By default, 8192. The list of all available parameters you can see in [MergeTreeSettings.h](https://github.com/yandex/ClickHouse/blob/master/dbms/src/Storages/MergeTree/MergeTreeSettings.h). + - `min_merge_bytes_to_use_direct_io` — The minimum data volume for merge operation required for using of the direct I/O access to the storage disk. During the merging of the data parts, ClickHouse calculates summary storage volume of all the data to be merged. If the volume exceeds `min_merge_bytes_to_use_direct_io` bytes, thеn ClickHouse reads and writes the data using direct I/O interface (`O_DIRECT` option) to the storage disk. If `min_merge_bytes_to_use_direct_io = 0`, then the direct I/O is disabled. Default value: `10 * 1024 * 1024 * 1024` bytes. + **Example of sections setting** diff --git a/docs/en/operations/troubleshooting.md b/docs/en/operations/troubleshooting.md index 445cd67bb7c..fdfff6b26b2 100644 --- a/docs/en/operations/troubleshooting.md +++ b/docs/en/operations/troubleshooting.md @@ -111,8 +111,8 @@ Check: Check: - - The `tcp_port_secure` setting. - - Settings for SSL sertificates. + - The [tcp_port_secure](server_settings/settings.md#server_settings-tcp_port_secure) setting. + - Settings for [SSL sertificates](server_settings/settings.md#server_settings-openssl). Use proper parameters while connecting. For example, use the `port_secure` parameter with `clickhouse_client`. diff --git a/docs/en/query_language/dicts/external_dicts_dict_sources.md b/docs/en/query_language/dicts/external_dicts_dict_sources.md index f26967c2d0f..8fb2145ecaf 100644 --- a/docs/en/query_language/dicts/external_dicts_dict_sources.md +++ b/docs/en/query_language/dicts/external_dicts_dict_sources.md @@ -88,7 +88,7 @@ Example of settings: ``` -In order for ClickHouse to access an HTTPS resource, you must [configure openSSL](../../operations/server_settings/settings.md) in the server configuration. +In order for ClickHouse to access an HTTPS resource, you must [configure openSSL](../../operations/server_settings/settings.md#server_settings-openssl) in the server configuration. Setting fields: diff --git a/docs/en/query_language/table_functions/remote.md b/docs/en/query_language/table_functions/remote.md index 06d9876f1c7..3b9263d959c 100644 --- a/docs/en/query_language/table_functions/remote.md +++ b/docs/en/query_language/table_functions/remote.md @@ -72,6 +72,6 @@ The `remote` table function can be useful in the following cases: If the user is not specified, `default` is used. If the password is not specified, an empty password is used. -`remoteSecure` - same as `remote` but with secured connection. Default port - `tcp_port_secure` from config or 9440. +`remoteSecure` - same as `remote` but with secured connection. Default port — [tcp_port_secure](../../operations/server_settings/settings.md#server_settings-tcp_port_secure) from config or 9440. [Original article](https://clickhouse.yandex/docs/en/query_language/table_functions/remote/) diff --git a/docs/ru/interfaces/formats.md b/docs/ru/interfaces/formats.md index 20cf1917c15..cd0fc9ae108 100644 --- a/docs/ru/interfaces/formats.md +++ b/docs/ru/interfaces/formats.md @@ -14,7 +14,6 @@ ClickHouse может принимать (`INSERT`) и отдавать (`SELECT [CSVWithNames](#csvwithnames) | ✔ | ✔ | [Values](#values) | ✔ | ✔ | [Vertical](#vertical) | ✗ | ✔ | -[VerticalRaw](#verticalraw) | ✗ | ✔ | [JSON](#json) | ✗ | ✔ | [JSONCompact](#jsoncompact) | ✗ | ✔ | [JSONEachRow](#jsoneachrow) | ✔ | ✔ | @@ -354,10 +353,22 @@ SELECT * FROM t_null └───┴──────┘ ``` +В форматах `Pretty*` строки выводятся без экранирования. Ниже приведен пример для формата [PrettyCompact](#prettycompact): + +``` sql +SELECT 'String with \'quotes\' and \t character' AS Escaping_test +``` + +``` +┌─Escaping_test────────────────────────┐ +│ String with 'quotes' and character │ +└──────────────────────────────────────┘ +``` + Для защиты от вываливания слишком большого количества данных в терминал, выводится только первые 10 000 строк. Если строк больше или равно 10 000, то будет написано "Showed first 10 000." Этот формат подходит только для вывода результата выполнения запроса, но не для парсинга (приёма данных для вставки в таблицу). -Формат Pretty поддерживает вывод тотальных значений (при использовании WITH TOTALS) и экстремальных значений (при настройке extremes выставленной в 1). В этих случаях, после основных данных выводятся тотальные значения, и экстремальные значения, в отдельных табличках. Пример (показан для формата PrettyCompact): +Формат `Pretty` поддерживает вывод тотальных значений (при использовании WITH TOTALS) и экстремальных значений (при настройке extremes выставленной в 1). В этих случаях, после основных данных выводятся тотальные значения, и экстремальные значения, в отдельных табличках. Пример (показан для формата [PrettyCompact](#prettycompact)): ``` sql SELECT EventDate, count() AS c FROM test.hits GROUP BY EventDate WITH TOTALS ORDER BY EventDate FORMAT PrettyCompact @@ -388,7 +399,7 @@ Extremes: ## PrettyCompact {#prettycompact} -Отличается от `Pretty` тем, что не рисуется сетка между строками - результат более компактный. +Отличается от [Pretty](#pretty) тем, что не рисуется сетка между строками - результат более компактный. Этот формат используется по умолчанию в клиенте командной строки в интерактивном режиме. ## PrettyCompactMonoBlock {#prettycompactmonoblock} @@ -433,6 +444,7 @@ FixedString представлены просто как последовате Array представлены как длина в формате varint (unsigned [LEB128](https://en.wikipedia.org/wiki/LEB128)), а затем элементы массива, подряд. Для поддержки [NULL](../query_language/syntax.md#null-literal) перед каждым значением типа [Nullable](../data_types/nullable.md + ## Values Выводит каждую строку в скобках. Строки разделены запятыми. После последней строки запятой нет. Значения внутри скобок также разделены запятыми. Числа выводятся в десятичном виде без кавычек. Массивы выводятся в квадратных скобках. Строки, даты, даты-с-временем выводятся в кавычках. Правила экранирования и особенности парсинга аналогичны формату [TabSeparated](#tabseparated). При форматировании, лишние пробелы не ставятся, а при парсинге - допустимы и пропускаются (за исключением пробелов внутри значений типа массив, которые недопустимы). [NULL](../query_language/syntax.md) представляется как `NULL`. @@ -459,34 +471,20 @@ x: 1 y: ᴺᵁᴸᴸ ``` -Этот формат подходит только для вывода результата выполнения запроса, но не для парсинга (приёма данных для вставки в таблицу). +В формате `Vertical` строки выводятся без экранирования. Например: -## VerticalRaw {#verticalraw} - -Отличается от формата `Vertical` тем, что строки выводятся без экранирования. -Этот формат подходит только для вывода результата выполнения запроса, но не для парсинга (приёма данных для вставки в таблицу). - -Примеры: +``` sql +SELECT 'string with \'quotes\' and \t with some special \n characters' AS test FORMAT Vertical ``` -:) SHOW CREATE TABLE geonames FORMAT VerticalRaw; -Row 1: -────── -statement: CREATE TABLE default.geonames ( geonameid UInt32, date Date DEFAULT CAST('2017-12-08' AS Date)) ENGINE = MergeTree(date, geonameid, 8192) -:) SELECT 'string with \'quotes\' and \t with some special \n characters' AS test FORMAT VerticalRaw; +``` Row 1: ────── -test: string with 'quotes' and with some special +test: string with 'quotes' and with some special characters ``` -Для сравнения - формат Vertical: -``` -:) SELECT 'string with \'quotes\' and \t with some special \n characters' AS test FORMAT Vertical; -Row 1: -────── -test: string with \'quotes\' and \t with some special \n characters -``` +Этот формат подходит только для вывода результата выполнения запроса, но не для парсинга (приёма данных для вставки в таблицу). ## XML {#xml}