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

Merge branch 'master' into keeper-storage

Browse Source
This commit is contained in:
Antonio Andelic 2022-08-31 10:12:11 +00:00
commit 9dd1a9859d
62 changed files with 617 additions and 1765 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
.gitmodules vendored
View File

@ -259,10 +259,6 @@
[submodule "contrib/minizip-ng"]
path = contrib/minizip-ng
url = https://github.com/zlib-ng/minizip-ng
[submodule "contrib/annoy"]
path = contrib/annoy
url = https://github.com/ClickHouse/annoy.git
branch = ClickHouse-master
[submodule "contrib/qpl"]
path = contrib/qpl
url = https://github.com/intel/qpl.git

22
base/glibc-compatibility/musl/dup3.c Normal file
View File

@ -0,0 +1,22 @@
#define _GNU_SOURCE
#include <unistd.h>
#include <errno.h>
#include <fcntl.h>
#include "syscall.h"
int dup3(int old, int new, int flags)
{
int r;
#ifdef SYS_dup2
if (old==new) return __syscall_ret(-EINVAL);
if (flags & O_CLOEXEC) {
while ((r=__syscall(SYS_dup3, old, new, flags))==-EBUSY);
if (r!=-ENOSYS) return __syscall_ret(r);
}
while ((r=__syscall(SYS_dup2, old, new))==-EBUSY);
if (flags & O_CLOEXEC) __syscall(SYS_fcntl, new, F_SETFD, FD_CLOEXEC);
#else
while ((r=__syscall(SYS_dup3, old, new, flags))==-EBUSY);
#endif
return __syscall_ret(r);
}

26
base/glibc-compatibility/musl/inotify.c Normal file
View File

@ -0,0 +1,26 @@
#include <sys/inotify.h>
#include <errno.h>
#include "syscall.h"
int inotify_init()
{
return inotify_init1(0);
}
int inotify_init1(int flags)
{
int r = __syscall(SYS_inotify_init1, flags);
#ifdef SYS_inotify_init
if (r==-ENOSYS && !flags) r = __syscall(SYS_inotify_init);
#endif
return __syscall_ret(r);
}
int inotify_add_watch(int fd, const char *pathname, uint32_t mask)
{
return syscall(SYS_inotify_add_watch, fd, pathname, mask);
}
int inotify_rm_watch(int fd, int wd)
{
return syscall(SYS_inotify_rm_watch, fd, wd);
}

2
contrib/CMakeLists.txt vendored
View File

@ -159,8 +159,6 @@ add_contrib (s2geometry-cmake s2geometry)
add_contrib (c-ares-cmake c-ares)
add_contrib (qpl-cmake qpl)
add_contrib(annoy-cmake annoy)
# Put all targets defined here and in subdirectories under "contrib/<immediate-subdir>" folders in GUI-based IDEs.
# Some of third-party projects may override CMAKE_FOLDER or FOLDER property of their targets, so they would not appear
# in "contrib/..." as originally planned, so we workaround this by fixing FOLDER properties of all targets manually,

1
contrib/annoy vendored

@ -1 +0,0 @@
Subproject commit 9d8a603a4cd252448589e84c9846f94368d5a289

16
contrib/annoy-cmake/CMakeLists.txt
View File

@ -1,16 +0,0 @@
option(ENABLE_ANNOY "Enable Annoy index support" ${ENABLE_LIBRARIES})
if ((NOT ENABLE_ANNOY) OR (SANITIZE STREQUAL "undefined"))
message (STATUS "Not using annoy")
return()
endif()
set(ANNOY_PROJECT_DIR "${ClickHouse_SOURCE_DIR}/contrib/annoy")
set(ANNOY_SOURCE_DIR "${ANNOY_PROJECT_DIR}/src")
add_library(_annoy INTERFACE)
target_include_directories(_annoy SYSTEM INTERFACE ${ANNOY_SOURCE_DIR})
add_library(ch_contrib::annoy ALIAS _annoy)
target_compile_definitions(_annoy INTERFACE ENABLE_ANNOY)
target_compile_definitions(_annoy INTERFACE ANNOYLIB_MULTITHREADED_BUILD)

2
contrib/libuv vendored

@ -1 +1 @@
Subproject commit 95081e7c16c9857babe6d4e2bc1c779198ea89ae
Subproject commit 3a85b2eb3d83f369b8a8cafd329d7e9dc28f60cf

6
contrib/libuv-cmake/CMakeLists.txt
View File

@ -15,6 +15,7 @@ set(uv_sources
src/inet.c
src/random.c
src/strscpy.c
src/strtok.c
src/threadpool.c
src/timer.c
src/uv-common.c
@ -75,13 +76,13 @@ if(CMAKE_SYSTEM_NAME STREQUAL "Linux")
list(APPEND uv_defines _GNU_SOURCE _POSIX_C_SOURCE=200112)
list(APPEND uv_libraries rt)
list(APPEND uv_sources
src/unix/epoll.c
src/unix/linux-core.c
src/unix/linux-inotify.c
src/unix/linux-syscalls.c
src/unix/procfs-exepath.c
src/unix/random-getrandom.c
src/unix/random-sysctl-linux.c
src/unix/sysinfo-loadavg.c)
src/unix/random-sysctl-linux.c)
endif()
if(CMAKE_SYSTEM_NAME STREQUAL "NetBSD")
@ -111,6 +112,7 @@ if(CMAKE_SYSTEM_NAME STREQUAL "OS/390")
src/unix/pthread-fixes.c
src/unix/pthread-barrier.c
src/unix/os390.c
src/unix/os390-proctitle.c
src/unix/os390-syscalls.c)
endif()

4
docs/en/engines/database-engines/mysql.md
View File

@ -4,7 +4,7 @@ sidebar_position: 50
sidebar_label: MySQL
---
# MySQL
# MySQL
Allows to connect to databases on a remote MySQL server and perform `INSERT` and `SELECT` queries to exchange data between ClickHouse and MySQL.
@ -99,7 +99,7 @@ mysql> select * from mysql_table;
Database in ClickHouse, exchanging data with the MySQL server:
``` sql
CREATE DATABASE mysql_db ENGINE = MySQL('localhost:3306', 'test', 'my_user', 'user_password')
CREATE DATABASE mysql_db ENGINE = MySQL('localhost:3306', 'test', 'my_user', 'user_password') SETTINGS read_write_timeout=10000, connect_timeout=100;
```
``` sql

125
docs/en/engines/table-engines/mergetree-family/annindexes.md
View File

@ -1,125 +0,0 @@
# Approximate Nearest Neighbor Search Indexes [experimental] {#table_engines-ANNIndex}
The main task that indexes achieve is to quickly find nearest neighbors for multidimensional data. An example of such a problem can be finding similar pictures (texts) for a given picture (text). That problem can be reduced to finding the nearest [embeddings](https://cloud.google.com/architecture/overview-extracting-and-serving-feature-embeddings-for-machine-learning). They can be created from data using [UDF](../../../sql-reference/functions/index.md#executable-user-defined-functions).
The next query finds the closest neighbors in N-dimensional space using the L2 (Euclidean) distance:
``` sql
SELECT *
FROM table_name
WHERE L2Distance(Column, Point) < MaxDistance
LIMIT N
```
But it will take some time for execution because of the long calculation of the distance between `TargetEmbedding` and all other vectors. This is where ANN indexes can help. They store a compact approximation of the search space (e.g. using clustering, search trees, etc.) and are able to compute approximate neighbors quickly.
## Indexes Structure
Approximate Nearest Neighbor Search Indexes (`ANNIndexes`) are similar to skip indexes. They are constructed by some granules and determine which of them should be skipped. Compared to skip indices, ANN indices use their results not only to skip some group of granules, but also to select particular granules from a set of granules.
`ANNIndexes` are designed to speed up two types of queries:
- ###### Type 1: Where
``` sql
SELECT *
FROM table_name
WHERE DistanceFunction(Column, Point) < MaxDistance
LIMIT N
```
- ###### Type 2: Order by
``` sql
SELECT *
FROM table_name [WHERE ...]
ORDER BY DistanceFunction(Column, Point)
LIMIT N
```
In these queries, `DistanceFunction` is selected from [distance functions](../../../sql-reference/functions/distance-functions). `Point` is a known vector (something like `(0.1, 0.1, ... )`). To avoid writing large vectors, use [client parameters](../../../interfaces/cli.md#queries-with-parameters-cli-queries-with-parameters). `Value` - a float value that will bound the neighbourhood.
!!! note "Note"
ANN index can't speed up query that satisfies both types(`where + order by`, only one of them). All queries must have the limit, as algorithms are used to find nearest neighbors and need a specific number of them.
!!! note "Note"
Indexes are applied only to queries with a limit less than the `max_limit_for_ann_queries` setting. This helps to avoid memory overflows in queries with a large limit. `max_limit_for_ann_queries` setting can be changed if you know you can provide enough memory. The default value is `1000000`.
Both types of queries are handled the same way. The indexes get `n` neighbors (where `n` is taken from the `LIMIT` clause) and work with them. In `ORDER BY` query they remember the numbers of all parts of the granule that have at least one of neighbor. In `WHERE` query they remember only those parts that satisfy the requirements.
## Create table with ANNIndex
```sql
CREATE TABLE t
(
`id` Int64,
`number` Tuple(Float32, Float32, Float32),
INDEX x number TYPE annoy GRANULARITY N
)
ENGINE = MergeTree
ORDER BY id;
```
```sql
CREATE TABLE t
(
`id` Int64,
`number` Array(Float32),
INDEX x number TYPE annoy GRANULARITY N
)
ENGINE = MergeTree
ORDER BY id;
```
With greater `GRANULARITY` indexes remember the data structure better. The `GRANULARITY` indicates how many granules will be used to construct the index. The more data is provided for the index, the more of it can be handled by one index and the more chances that with the right hyperparameters the index will remember the data structure better. But some indexes can't be built if they don't have enough data, so this granule will always participate in the query. For more information, see the description of indexes.
As the indexes are built only during insertions into table, `INSERT` and `OPTIMIZE` queries are slower than for ordinary table. At this stage indexes remember all the information about the given data. ANNIndexes should be used if you have immutable or rarely changed data and many read requests.
You can create your table with index which uses certain algorithm. Now only indices based on the following algorithms are supported:
# Index list
- [Annoy](../../../engines/table-engines/mergetree-family/annindexes.md#annoy-annoy)
# Annoy {#annoy}
Implementation of the algorithm was taken from [this repository](https://github.com/spotify/annoy).
Short description of the algorithm:
The algorithm recursively divides in half all space by random linear surfaces (lines in 2D, planes in 3D e.t.c.). Thus it makes tree of polyhedrons and points that they contains. Repeating the operation several times for greater accuracy it creates a forest.
To find K Nearest Neighbours it goes down through the trees and fills the buffer of closest points using the priority queue of polyhedrons. Next, it sorts buffer and return the nearest K points.
__Examples__:
```sql
CREATE TABLE t
(
id Int64,
number Tuple(Float32, Float32, Float32),
INDEX x number TYPE annoy(T) GRANULARITY N
)
ENGINE = MergeTree
ORDER BY id;
```
```sql
CREATE TABLE t
(
id Int64,
number Array(Float32),
INDEX x number TYPE annoy(T) GRANULARITY N
)
ENGINE = MergeTree
ORDER BY id;
```
!!! note "Note"
Table with array field will work faster, but all arrays **must** have same length. Use [CONSTRAINT](../../../sql-reference/statements/create/table.md#constraints) to avoid errors. For example, `CONSTRAINT constraint_name_1 CHECK length(number) = 256`.
Parameter `T` is the number of trees which algorithm will create. The bigger it is, the slower (approximately linear) it works (in both `CREATE` and `SELECT` requests), but the better accuracy you get (adjusted for randomness).
Annoy supports only `L2Distance`.
In the `SELECT` in the settings (`ann_index_select_query_params`) you can specify the size of the internal buffer (more details in the description above or in the [original repository](https://github.com/spotify/annoy)). During the query it will inspect up to `search_k` nodes which defaults to `n_trees * n` if not provided. `search_k` gives you a run-time tradeoff between better accuracy and speed.
__Example__:
``` sql
SELECT *
FROM table_name [WHERE ...]
ORDER BY L2Distance(Column, Point)
LIMIT N
SETTING ann_index_select_query_params=`k_search=100`
```

4
docs/en/engines/table-engines/mergetree-family/mergetree.md
View File

@ -481,10 +481,6 @@ For example:
- `NOT startsWith(s, 'test')`
:::
## Approximate Nearest Neighbor Search Indexes [experimental] {#table_engines-ANNIndex}
In addition to skip indices, there are also [Approximate Nearest Neighbor Search Indexes](../../../engines/table-engines/mergetree-family/annindexes.md).
## Projections {#projections}
Projections are like [materialized views](../../../sql-reference/statements/create/view.md#materialized) but defined in part-level. It provides consistency guarantees along with automatic usage in queries.

2
docs/en/sql-reference/data-types/decimal.md
View File

@ -46,7 +46,7 @@ Binary operations on Decimal result in wider result type (with any order of argu
Rules for scale:
- add, subtract: S = max(S1, S2).
- multuply: S = S1 + S2.
- multiply: S = S1 + S2.
- divide: S = S1.
For similar operations between Decimal and integers, the result is Decimal of the same size as an argument.

57
docs/en/sql-reference/functions/string-functions.md
View File

@ -495,25 +495,23 @@ If the ‘s’ string is non-empty and does not contain the ‘c’ character at
Returns the string ‘s’ that was converted from the encoding in ‘from’ to the encoding in ‘to’.
## Base58Encode(plaintext), Base58Decode(encoded_text)
## base58Encode(plaintext)
Accepts a String and encodes/decodes it using [Base58](https://tools.ietf.org/id/draft-msporny-base58-01.html) encoding scheme using "Bitcoin" alphabet.
Accepts a String and encodes it using [Base58](https://tools.ietf.org/id/draft-msporny-base58-01.html) encoding scheme using "Bitcoin" alphabet.
**Syntax**
```sql
base58Encode(decoded)
base58Decode(encoded)
base58Encode(plaintext)
```
**Arguments**
- `decoded` — [String](../../sql-reference/data-types/string.md) column or constant.
- `encoded` — [String](../../sql-reference/data-types/string.md) column or constant. If the string is not a valid base58-encoded value, an exception is thrown.
- `plaintext` — [String](../../sql-reference/data-types/string.md) column or constant.
**Returned value**
- A string containing encoded/decoded value of 1st argument.
- A string containing encoded value of 1st argument.
Type: [String](../../sql-reference/data-types/string.md).
@ -523,17 +521,48 @@ Query:
``` sql
SELECT base58Encode('Encoded');
SELECT base58Encode('3dc8KtHrwM');
```
Result:
```text
┌─encodeBase58('Encoded')─┐
│ 3dc8KtHrwM │
└──────────────────────────────────┘
┌─decodeBase58('3dc8KtHrwM')─┐
│ Encoded │
└────────────────────────────────────┘
┌─base58Encode('Encoded')─┐
│ 3dc8KtHrwM │
└─────────────────────────┘
```
## base58Decode(encoded_text)
Accepts a String and decodes it using [Base58](https://tools.ietf.org/id/draft-msporny-base58-01.html) encoding scheme using "Bitcoin" alphabet.
**Syntax**
```sql
base58Decode(encoded_text)
```
**Arguments**
- `encoded_text` — [String](../../sql-reference/data-types/string.md) column or constant. If the string is not a valid base58-encoded value, an exception is thrown.
**Returned value**
- A string containing decoded value of 1st argument.
Type: [String](../../sql-reference/data-types/string.md).
**Example**
Query:
``` sql
SELECT base58Decode('3dc8KtHrwM');
```
Result:
```text
┌─base58Decode('3dc8KtHrwM')─┐
│ Encoded │
└────────────────────────────┘
```
## base64Encode(s)

28
docs/ru/sql-reference/functions/string-functions.md
View File

@ -16,7 +16,7 @@ sidebar_label: "Функции для работы со строками"
empty(x)
```
Строка считается непустой, если содержит хотя бы один байт, пусть даже это пробел или нулевой байт.
Строка считается непустой, если содержит хотя бы один байт, пусть даже это пробел или нулевой байт.
Функция также поддерживает работу с типами [Array](array-functions.md#function-empty) и [UUID](uuid-functions.md#empty).
@ -56,7 +56,7 @@ SELECT empty('text');
notEmpty(x)
```
Строка считается непустой, если содержит хотя бы один байт, пусть даже это пробел или нулевой байт.
Строка считается непустой, если содержит хотя бы один байт, пусть даже это пробел или нулевой байт.
Функция также поддерживает работу с типами [Array](array-functions.md#function-notempty) и [UUID](uuid-functions.md#notempty).
@ -491,21 +491,21 @@ SELECT concat(key1, key2), sum(value) FROM key_val GROUP BY (key1, key2);
Возвращает сконвертированную из кодировки from в кодировку to строку s.
## Base58Encode(plaintext), Base58Decode(encoded_text) {#base58}
## base58Encode(plaintext), base58Decode(encoded_text) {#base58}
Принимает на вход строку или колонку строк и кодирует/раскодирует их с помощью схемы кодирования [Base58](https://tools.ietf.org/id/draft-msporny-base58-01.html) с использованием стандартного алфавита Bitcoin.
**Синтаксис**
```sql
encodeBase58(decoded)
decodeBase58(encoded)
base58Encode(decoded)
base58Decode(encoded)
```
**Аргументы**
- `decoded` — Колонка или строка типа [String](../../sql-reference/data-types/string.md).
- `encoded` — Колонка или строка типа [String](../../sql-reference/data-types/string.md). Если входная строка не является корректным кодом для какой-либо другой строки, возникнет исключение `1001`.
- `encoded` — Колонка или строка типа [String](../../sql-reference/data-types/string.md). Если входная строка не является корректным кодом для какой-либо другой строки, возникнет исключение.
**Возвращаемое значение**
@ -518,18 +518,18 @@ decodeBase58(encoded)
Запрос:
``` sql
SELECT encodeBase58('encode');
SELECT decodeBase58('izCFiDUY');
SELECT base58Encode('Encoded');
SELECT base58Decode('3dc8KtHrwM');
```
Результат:
```text
┌─encodeBase58('encode', 'flickr')─┐
│ SvyTHb1D │
└──────────────────────────────────┘
┌─decodeBase58('izCFiDUY', 'ripple')─┐
│ decode │
└────────────────────────────────────┘
┌─base58Encode('Encoded')─┐
│ 3dc8KtHrwM │
└─────────────────────────┘
┌─base58Decode('3dc8KtHrwM')─┐
│ Encoded │
└────────────────────────────┘
```
## base64Encode(s) {#base64encode}

4
src/Access/Common/AllowedClientHosts.cpp
View File

@ -110,7 +110,7 @@ namespace
}
/// Returns the host name by its address.
Strings getHostsByAddress(const IPAddress & address)
std::unordered_set<String> getHostsByAddress(const IPAddress & address)
{
auto hosts = DNSResolver::instance().reverseResolve(address);
@ -526,7 +526,7 @@ bool AllowedClientHosts::contains(const IPAddress & client_address) const
return true;
/// Check `name_regexps`.
std::optional<Strings> resolved_hosts;
std::optional<std::unordered_set<String>> resolved_hosts;
auto check_name_regexp = [&](const String & name_regexp_)
{
try

4
src/CMakeLists.txt
View File

@ -570,10 +570,6 @@ endif()
dbms_target_link_libraries(PUBLIC ch_contrib::consistent_hashing)
if (TARGET ch_contrib::annoy)
dbms_target_link_libraries(PUBLIC ch_contrib::annoy)
endif()
include ("${ClickHouse_SOURCE_DIR}/cmake/add_check.cmake")
if (ENABLE_TESTS)

26
src/Common/CaresPTRResolver.cpp
View File

@ -15,13 +15,23 @@ namespace DB
static void callback(void * arg, int status, int, struct hostent * host)
{
auto * ptr_records = reinterpret_cast<std::vector<std::string>*>(arg);
auto * ptr_records = reinterpret_cast<std::unordered_set<std::string>*>(arg);
if (status == ARES_SUCCESS && host->h_aliases)
{
/*
* In some cases (e.g /etc/hosts), hostent::h_name is filled and hostent::h_aliases is empty.
* Thus, we can't rely solely on hostent::h_aliases. More info on:
* https://github.com/ClickHouse/ClickHouse/issues/40595#issuecomment-1230526931
* */
if (auto * ptr_record = host->h_name)
{
ptr_records->insert(ptr_record);
}
int i = 0;
while (auto * ptr_record = host->h_aliases[i])
{
ptr_records->emplace_back(ptr_record);
ptr_records->insert(ptr_record);
i++;
}
}
@ -58,9 +68,9 @@ namespace DB
* */
}
std::vector<std::string> CaresPTRResolver::resolve(const std::string & ip)
std::unordered_set<std::string> CaresPTRResolver::resolve(const std::string & ip)
{
std::vector<std::string> ptr_records;
std::unordered_set<std::string> ptr_records;
resolve(ip, ptr_records);
wait();
@ -68,9 +78,9 @@ namespace DB
return ptr_records;
}
std::vector<std::string> CaresPTRResolver::resolve_v6(const std::string & ip)
std::unordered_set<std::string> CaresPTRResolver::resolve_v6(const std::string & ip)
{
std::vector<std::string> ptr_records;
std::unordered_set<std::string> ptr_records;
resolve_v6(ip, ptr_records);
wait();
@ -78,7 +88,7 @@ namespace DB
return ptr_records;
}
void CaresPTRResolver::resolve(const std::string & ip, std::vector<std::string> & response)
void CaresPTRResolver::resolve(const std::string & ip, std::unordered_set<std::string> & response)
{
in_addr addr;
@ -87,7 +97,7 @@ namespace DB
ares_gethostbyaddr(channel, reinterpret_cast<const void*>(&addr), sizeof(addr), AF_INET, callback, &response);
}
void CaresPTRResolver::resolve_v6(const std::string & ip, std::vector<std::string> & response)
void CaresPTRResolver::resolve_v6(const std::string & ip, std::unordered_set<std::string> & response)
{
in6_addr addr;
inet_pton(AF_INET6, ip.c_str(), &addr);

8
src/Common/CaresPTRResolver.h
View File

@ -25,16 +25,16 @@ namespace DB
explicit CaresPTRResolver(provider_token);
~CaresPTRResolver() override;
std::vector<std::string> resolve(const std::string & ip) override;
std::unordered_set<std::string> resolve(const std::string & ip) override;
std::vector<std::string> resolve_v6(const std::string & ip) override;
std::unordered_set<std::string> resolve_v6(const std::string & ip) override;
private:
void wait();
void resolve(const std::string & ip, std::vector<std::string> & response);
void resolve(const std::string & ip, std::unordered_set<std::string> & response);
void resolve_v6(const std::string & ip, std::vector<std::string> & response);
void resolve_v6(const std::string & ip, std::unordered_set<std::string> & response);
ares_channel channel;
};

6
src/Common/DNSPTRResolver.h
View File

@ -1,7 +1,7 @@
#pragma once
#include <string>
#include <vector>
#include <unordered_set>
namespace DB
{
@ -10,9 +10,9 @@ namespace DB
virtual ~DNSPTRResolver() = default;
virtual std::vector<std::string> resolve(const std::string & ip) = 0;
virtual std::unordered_set<std::string> resolve(const std::string & ip) = 0;
virtual std::vector<std::string> resolve_v6(const std::string & ip) = 0;
virtual std::unordered_set<std::string> resolve_v6(const std::string & ip) = 0;
};
}

4
src/Common/DNSResolver.cpp
View File

@ -136,7 +136,7 @@ static DNSResolver::IPAddresses resolveIPAddressImpl(const std::string & host)
return addresses;
}
static Strings reverseResolveImpl(const Poco::Net::IPAddress & address)
static std::unordered_set<String> reverseResolveImpl(const Poco::Net::IPAddress & address)
{
auto ptr_resolver = DB::DNSPTRResolverProvider::get();
@ -234,7 +234,7 @@ std::vector<Poco::Net::SocketAddress> DNSResolver::resolveAddressList(const std:
return addresses;
}
Strings DNSResolver::reverseResolve(const Poco::Net::IPAddress & address)
std::unordered_set<String> DNSResolver::reverseResolve(const Poco::Net::IPAddress & address)
{
if (impl->disable_cache)
return reverseResolveImpl(address);

2
src/Common/DNSResolver.h
View File

@ -37,7 +37,7 @@ public:
std::vector<Poco::Net::SocketAddress> resolveAddressList(const std::string & host, UInt16 port);
/// Accepts host IP and resolves its host names
Strings reverseResolve(const Poco::Net::IPAddress & address);
std::unordered_set<String> reverseResolve(const Poco::Net::IPAddress & address);
/// Get this server host name
String getHostName();

122
src/Common/FileCache.cpp
View File

@ -71,10 +71,15 @@ bool FileCache::isReadOnly()
return !isQueryInitialized();
}
void FileCache::assertInitialized() const
void FileCache::assertInitialized(std::lock_guard<std::mutex> & /* cache_lock */) const
{
if (!is_initialized)
throw Exception(ErrorCodes::REMOTE_FS_OBJECT_CACHE_ERROR, "Cache not initialized");
{
if (initialization_exception)
std::rethrow_exception(initialization_exception);
else
throw Exception(ErrorCodes::REMOTE_FS_OBJECT_CACHE_ERROR, "Cache not initialized");
}
}
void FileCache::initialize()
@ -90,28 +95,43 @@ void FileCache::initialize()
}
catch (...)
{
tryLogCurrentException(__PRETTY_FUNCTION__);
initialization_exception = std::current_exception();
throw;
}
}
else
{
fs::create_directories(cache_base_path);
}
is_initialized = true;
}
}
void FileCache::useCell(
const FileSegmentCell & cell, FileSegments & result, std::lock_guard<std::mutex> & cache_lock) const
const FileSegmentCell & cell, FileSegments & result, std::lock_guard<std::mutex> & cache_lock)
{
auto file_segment = cell.file_segment;
if (file_segment->isDownloaded()
&& fs::file_size(getPathInLocalCache(file_segment->key(), file_segment->offset(), file_segment->isPersistent())) == 0)
throw Exception(
ErrorCodes::LOGICAL_ERROR,
"Cannot have zero size downloaded file segments. {}",
file_segment->getInfoForLog());
if (file_segment->isDownloaded())
{
fs::path path = file_segment->getPathInLocalCache();
if (!fs::exists(path))
{
throw Exception(
ErrorCodes::LOGICAL_ERROR,
"File path does not exist, but file has DOWNLOADED state. {}",
file_segment->getInfoForLog());
}
if (fs::file_size(path) == 0)
{
throw Exception(
ErrorCodes::LOGICAL_ERROR,
"Cannot have zero size downloaded file segments. {}",
file_segment->getInfoForLog());
}
}
result.push_back(cell.file_segment);
@ -222,7 +242,12 @@ FileSegments FileCache::getImpl(
}
FileSegments FileCache::splitRangeIntoCells(
const Key & key, size_t offset, size_t size, FileSegment::State state, bool is_persistent, std::lock_guard<std::mutex> & cache_lock)
const Key & key,
size_t offset,
size_t size,
FileSegment::State state,
bool is_persistent,
std::lock_guard<std::mutex> & cache_lock)
{
assert(size > 0);
@ -346,16 +371,16 @@ void FileCache::fillHolesWithEmptyFileSegments(
FileSegmentsHolder FileCache::getOrSet(const Key & key, size_t offset, size_t size, bool is_persistent)
{
assertInitialized();
FileSegment::Range range(offset, offset + size - 1);
std::lock_guard cache_lock(mutex);
assertInitialized(cache_lock);
#ifndef NDEBUG
assertCacheCorrectness(key, cache_lock);
#endif
FileSegment::Range range(offset, offset + size - 1);
/// Get all segments which intersect with the given range.
auto file_segments = getImpl(key, range, cache_lock);
@ -374,16 +399,16 @@ FileSegmentsHolder FileCache::getOrSet(const Key & key, size_t offset, size_t si
FileSegmentsHolder FileCache::get(const Key & key, size_t offset, size_t size)
{
assertInitialized();
FileSegment::Range range(offset, offset + size - 1);
std::lock_guard cache_lock(mutex);
assertInitialized(cache_lock);
#ifndef NDEBUG
assertCacheCorrectness(key, cache_lock);
#endif
FileSegment::Range range(offset, offset + size - 1);
/// Get all segments which intersect with the given range.
auto file_segments = getImpl(key, range, cache_lock);
@ -417,7 +442,7 @@ FileCache::FileSegmentCell * FileCache::addCell(
if (files[key].contains(offset))
throw Exception(
ErrorCodes::LOGICAL_ERROR,
"Cache already exists for key: `{}`, offset: {}, size: {}.\nCurrent cache structure: {}",
"Cache cell already exists for key: `{}`, offset: {}, size: {}.\nCurrent cache structure: {}",
key.toString(), offset, size, dumpStructureUnlocked(key, cache_lock));
auto skip_or_download = [&]() -> FileSegmentPtr
@ -605,9 +630,7 @@ bool FileCache::tryReserve(const Key & key, size_t offset, size_t size, std::loc
auto remove_file_segment = [&](FileSegmentPtr file_segment, size_t file_segment_size)
{
query_context->remove(file_segment->key(), file_segment->offset(), file_segment_size, cache_lock);
std::lock_guard segment_lock(file_segment->mutex);
remove(file_segment->key(), file_segment->offset(), cache_lock, segment_lock);
remove(file_segment, cache_lock);
};
assert(trash.empty());
@ -724,19 +747,13 @@ bool FileCache::tryReserveForMainList(
}
}
auto remove_file_segment = [&](FileSegmentPtr file_segment)
{
std::lock_guard segment_lock(file_segment->mutex);
remove(file_segment->key(), file_segment->offset(), cache_lock, segment_lock);
};
/// This case is very unlikely, can happen in case of exception from
/// file_segment->complete(), which would be a logical error.
assert(trash.empty());
for (auto & cell : trash)
{
if (auto file_segment = cell->file_segment)
remove_file_segment(file_segment);
remove(file_segment, cache_lock);
}
if (is_overflow())
@ -758,7 +775,7 @@ bool FileCache::tryReserveForMainList(
for (auto & cell : to_evict)
{
if (auto file_segment = cell->file_segment)
remove_file_segment(file_segment);
remove(file_segment, cache_lock);
}
if (main_priority->getCacheSize(cache_lock) > (1ull << 63))
@ -772,10 +789,10 @@ bool FileCache::tryReserveForMainList(
void FileCache::removeIfExists(const Key & key)
{
assertInitialized();
std::lock_guard cache_lock(mutex);
assertInitialized(cache_lock);
auto it = files.find(key);
if (it == files.end())
return;
@ -869,27 +886,36 @@ void FileCache::removeIfReleasable()
#endif
}
void FileCache::remove(FileSegmentPtr file_segment, std::lock_guard<std::mutex> & cache_lock)
{
std::lock_guard segment_lock(file_segment->mutex);
remove(file_segment->key(), file_segment->offset(), cache_lock, segment_lock);
}
void FileCache::remove(
Key key, size_t offset,
std::lock_guard<std::mutex> & cache_lock, std::lock_guard<std::mutex> & /* segment_lock */)
{
LOG_DEBUG(log, "Remove from cache. Key: {}, offset: {}", key.toString(), offset);
auto * cell = getCell(key, offset, cache_lock);
if (!cell)
throw Exception(ErrorCodes::LOGICAL_ERROR, "No cache cell for key: {}, offset: {}", key.toString(), offset);
String cache_file_path;
bool is_persistent_file_segment = cell->file_segment->isPersistent();
if (cell->queue_iterator)
{
cell->queue_iterator->removeAndGetNext(cache_lock);
auto * cell = getCell(key, offset, cache_lock);
if (!cell)
throw Exception(ErrorCodes::LOGICAL_ERROR, "No cache cell for key: {}, offset: {}", key.toString(), offset);
if (cell->queue_iterator)
{
cell->queue_iterator->removeAndGetNext(cache_lock);
}
cache_file_path = cell->file_segment->getPathInLocalCache();
}
auto & offsets = files[key];
offsets.erase(offset);
auto cache_file_path = getPathInLocalCache(key, offset, is_persistent_file_segment);
if (fs::exists(cache_file_path))
{
try
@ -908,9 +934,10 @@ void FileCache::remove(
}
catch (...)
{
throw Exception(ErrorCodes::REMOTE_FS_OBJECT_CACHE_ERROR,
"Removal of cached file failed. Key: {}, offset: {}, path: {}, error: {}",
key.toString(), offset, cache_file_path, getCurrentExceptionMessage(false));
throw Exception(
ErrorCodes::REMOTE_FS_OBJECT_CACHE_ERROR,
"Removal of cached file failed. Key: {}, offset: {}, path: {}, error: {}",
key.toString(), offset, cache_file_path, getCurrentExceptionMessage(false));
}
}
}
@ -1139,9 +1166,10 @@ FileCache::FileSegmentCell::FileSegmentCell(
break;
}
default:
throw Exception(ErrorCodes::REMOTE_FS_OBJECT_CACHE_ERROR,
"Can create cell with either EMPTY, DOWNLOADED, DOWNLOADING state, got: {}",
FileSegment::stateToString(file_segment->download_state));
throw Exception(
ErrorCodes::REMOTE_FS_OBJECT_CACHE_ERROR,
"Can create cell with either EMPTY, DOWNLOADED, DOWNLOADING state, got: {}",
FileSegment::stateToString(file_segment->download_state));
}
}

10
src/Common/FileCache.h
View File

@ -140,7 +140,9 @@ private:
bool enable_filesystem_query_cache_limit;
Poco::Logger * log;
bool is_initialized = false;
std::exception_ptr initialization_exception;
mutable std::mutex mutex;
@ -152,6 +154,10 @@ private:
std::lock_guard<std::mutex> & cache_lock,
std::lock_guard<std::mutex> & segment_lock);
void remove(
FileSegmentPtr file_segment,
std::lock_guard<std::mutex> & cache_lock);
bool isLastFileSegmentHolder(
const Key & key,
size_t offset,
@ -164,7 +170,7 @@ private:
std::lock_guard<std::mutex> & cache_lock,
std::lock_guard<std::mutex> & segment_lock);
void assertInitialized() const;
void assertInitialized(std::lock_guard<std::mutex> & cache_lock) const;
struct FileSegmentCell : private boost::noncopyable
{
@ -220,7 +226,7 @@ private:
bool is_persistent,
std::lock_guard<std::mutex> & cache_lock);
void useCell(const FileSegmentCell & cell, FileSegments & result, std::lock_guard<std::mutex> & cache_lock) const;
static void useCell(const FileSegmentCell & cell, FileSegments & result, std::lock_guard<std::mutex> & cache_lock);
bool tryReserveForMainList(
const Key & key,

110
src/Common/FileSegment.cpp
View File

@ -78,19 +78,19 @@ FileSegment::State FileSegment::state() const
size_t FileSegment::getDownloadOffset() const
{
std::lock_guard segment_lock(mutex);
return range().left + getDownloadedSize(segment_lock);
return range().left + getDownloadedSizeUnlocked(segment_lock);
}
size_t FileSegment::getDownloadedSize() const
{
std::lock_guard segment_lock(mutex);
return getDownloadedSize(segment_lock);
return getDownloadedSizeUnlocked(segment_lock);
}
size_t FileSegment::getRemainingSizeToDownload() const
{
std::lock_guard segment_lock(mutex);
return range().size() - downloaded_size;
return range().size() - getDownloadedSizeUnlocked(segment_lock);
}
bool FileSegment::isDetached() const
@ -99,7 +99,7 @@ bool FileSegment::isDetached() const
return is_detached;
}
size_t FileSegment::getDownloadedSize(std::lock_guard<std::mutex> & /* segment_lock */) const
size_t FileSegment::getDownloadedSizeUnlocked(std::lock_guard<std::mutex> & /* segment_lock */) const
{
if (download_state == State::DOWNLOADED)
return downloaded_size;
@ -159,7 +159,7 @@ void FileSegment::resetDownloader()
void FileSegment::resetDownloaderImpl(std::lock_guard<std::mutex> & segment_lock)
{
if (downloaded_size == range().size())
if (getDownloadedSizeUnlocked(segment_lock) == range().size())
setDownloaded(segment_lock);
else
download_state = State::PARTIALLY_DOWNLOADED;
@ -241,14 +241,16 @@ void FileSegment::write(const char * from, size_t size, size_t offset_)
"Not enough space is reserved. Available: {}, expected: {}", availableSize(), size);
if (!isDownloader())
throw Exception(ErrorCodes::LOGICAL_ERROR,
"Only downloader can do the downloading. (CallerId: {}, DownloaderId: {})",
getCallerId(), downloader_id);
throw Exception(
ErrorCodes::LOGICAL_ERROR,
"Only downloader can do the downloading. (CallerId: {}, DownloaderId: {})",
getCallerId(), downloader_id);
if (downloaded_size == range().size())
throw Exception(ErrorCodes::REMOTE_FS_OBJECT_CACHE_ERROR,
"Attempt to write {} bytes to offset: {}, but current file segment is already fully downloaded",
size, offset_);
if (getDownloadedSize() == range().size())
throw Exception(
ErrorCodes::REMOTE_FS_OBJECT_CACHE_ERROR,
"Attempt to write {} bytes to offset: {}, but current file segment is already fully downloaded",
size, offset_);
auto download_offset = range().left + downloaded_size;
if (offset_ != download_offset)
@ -331,9 +333,9 @@ FileSegment::State FileSegment::wait()
return download_state;
}
bool FileSegment::reserve(size_t size)
bool FileSegment::reserve(size_t size_to_reserve)
{
if (!size)
if (!size_to_reserve)
throw Exception(ErrorCodes::REMOTE_FS_OBJECT_CACHE_ERROR, "Zero space reservation is not allowed");
{
@ -350,12 +352,16 @@ bool FileSegment::reserve(size_t size)
caller_id, downloader_id);
}
if (downloaded_size + size > range().size())
throw Exception(ErrorCodes::REMOTE_FS_OBJECT_CACHE_ERROR,
"Attempt to reserve space too much space ({}) for file segment with range: {} (downloaded size: {})",
size, range().toString(), downloaded_size);
size_t current_downloaded_size = getDownloadedSizeUnlocked(segment_lock);
if (current_downloaded_size + size_to_reserve > range().size())
{
throw Exception(
ErrorCodes::REMOTE_FS_OBJECT_CACHE_ERROR,
"Attempt to reserve space too much space: {} ({})",
size_to_reserve, getInfoForLogImpl(segment_lock));
}
assert(reserved_size >= downloaded_size);
assert(reserved_size >= current_downloaded_size);
}
/**
@ -363,29 +369,45 @@ bool FileSegment::reserve(size_t size)
* in case previous downloader did not fully download current file_segment
* and the caller is going to continue;
*/
size_t free_space = reserved_size - downloaded_size;
size_t size_to_reserve = size - free_space;
std::lock_guard cache_lock(cache->mutex);
size_t current_downloaded_size = getDownloadedSize();
assert(reserved_size >= current_downloaded_size);
size_t already_reserved_size = reserved_size - current_downloaded_size;
bool reserved = cache->tryReserve(key(), offset(), size_to_reserve, cache_lock);
if (reserved)
bool reserved = already_reserved_size >= size_to_reserve;
if (!reserved)
{
std::lock_guard segment_lock(mutex);
reserved_size += size;
std::lock_guard cache_lock(cache->mutex);
size_to_reserve = size_to_reserve - already_reserved_size;
reserved = cache->tryReserve(key(), offset(), size_to_reserve, cache_lock);
if (reserved)
{
std::lock_guard segment_lock(mutex);
reserved_size += size_to_reserve;
}
}
return reserved;
}
void FileSegment::setDownloaded(std::lock_guard<std::mutex> & /* segment_lock */)
bool FileSegment::isDownloaded() const
{
std::lock_guard segment_lock(mutex);
return isDownloadedUnlocked(segment_lock);
}
bool FileSegment::isDownloadedUnlocked(std::lock_guard<std::mutex> & /* segment_lock */) const
{
return is_downloaded;
}
void FileSegment::setDownloaded([[maybe_unused]] std::lock_guard<std::mutex> & segment_lock)
{
if (is_downloaded)
return;
download_state = State::DOWNLOADED;
is_downloaded = true;
downloader_id.clear();
if (cache_writer)
@ -394,6 +416,12 @@ void FileSegment::setDownloaded(std::lock_guard<std::mutex> & /* segment_lock */
cache_writer.reset();
remote_file_reader.reset();
}
download_state = State::DOWNLOADED;
is_downloaded = true;
assert(getDownloadedSizeUnlocked(segment_lock) > 0);
assert(std::filesystem::file_size(getPathInLocalCache()) > 0);
}
void FileSegment::setDownloadFailed(std::lock_guard<std::mutex> & /* segment_lock */)
@ -426,7 +454,7 @@ void FileSegment::completeBatchAndResetDownloader()
resetDownloaderImpl(segment_lock);
LOG_TEST(log, "Complete batch. Current downloaded size: {}", downloaded_size);
LOG_TEST(log, "Complete batch. Current downloaded size: {}", getDownloadedSizeUnlocked(segment_lock));
cv.notify_all();
}
@ -475,7 +503,7 @@ void FileSegment::completeBasedOnCurrentState(std::lock_guard<std::mutex> & cach
bool is_downloader = isDownloaderImpl(segment_lock);
bool is_last_holder = cache->isLastFileSegmentHolder(key(), offset(), cache_lock, segment_lock);
bool can_update_segment_state = is_downloader || is_last_holder;
size_t current_downloaded_size = getDownloadedSize(segment_lock);
size_t current_downloaded_size = getDownloadedSizeUnlocked(segment_lock);
SCOPE_EXIT({
if (is_downloader)
@ -494,13 +522,6 @@ void FileSegment::completeBasedOnCurrentState(std::lock_guard<std::mutex> & cach
download_state = State::PARTIALLY_DOWNLOADED;
resetDownloaderImpl(segment_lock);
if (cache_writer)
{
cache_writer->finalize();
cache_writer.reset();
remote_file_reader.reset();
}
}
switch (download_state)
@ -514,8 +535,8 @@ void FileSegment::completeBasedOnCurrentState(std::lock_guard<std::mutex> & cach
}
case State::DOWNLOADED:
{
assert(downloaded_size == range().size());
assert(is_downloaded);
assert(getDownloadedSizeUnlocked(segment_lock) == range().size());
assert(isDownloadedUnlocked(segment_lock));
break;
}
case State::DOWNLOADING:
@ -577,7 +598,7 @@ String FileSegment::getInfoForLogImpl(std::lock_guard<std::mutex> & segment_lock
info << "File segment: " << range().toString() << ", ";
info << "key: " << key().toString() << ", ";
info << "state: " << download_state << ", ";
info << "downloaded size: " << getDownloadedSize(segment_lock) << ", ";
info << "downloaded size: " << getDownloadedSizeUnlocked(segment_lock) << ", ";
info << "reserved size: " << reserved_size << ", ";
info << "downloader id: " << (downloader_id.empty() ? "None" : downloader_id) << ", ";
info << "caller id: " << getCallerId() << ", ";
@ -673,7 +694,7 @@ FileSegmentPtr FileSegment::getSnapshot(const FileSegmentPtr & file_segment, std
snapshot->hits_count = file_segment->getHitsCount();
snapshot->ref_count = file_segment.use_count();
snapshot->downloaded_size = file_segment->getDownloadedSize(segment_lock);
snapshot->downloaded_size = file_segment->getDownloadedSizeUnlocked(segment_lock);
snapshot->download_state = file_segment->download_state;
snapshot->is_persistent = file_segment->isPersistent();
@ -818,7 +839,8 @@ void FileSegmentRangeWriter::completeFileSegment(FileSegment & file_segment)
if (file_segment.isDetached())
return;
if (file_segment.getDownloadedSize() > 0)
size_t current_downloaded_size = file_segment.getDownloadedSize();
if (current_downloaded_size > 0)
{
file_segment.getOrSetDownloader();

9
src/Common/FileSegment.h
View File

@ -130,7 +130,7 @@ public:
bool isDownloader() const;
bool isDownloaded() const { return is_downloaded.load(); }
bool isDownloaded() const;
static String getCallerId();
@ -171,7 +171,7 @@ public:
private:
size_t availableSize() const { return reserved_size - downloaded_size; }
size_t getDownloadedSize(std::lock_guard<std::mutex> & segment_lock) const;
size_t getDownloadedSizeUnlocked(std::lock_guard<std::mutex> & segment_lock) const;
String getInfoForLogImpl(std::lock_guard<std::mutex> & segment_lock) const;
void assertCorrectnessImpl(std::lock_guard<std::mutex> & segment_lock) const;
bool hasFinalizedState() const;
@ -187,6 +187,8 @@ private:
void setDownloadFailed(std::lock_guard<std::mutex> & segment_lock);
bool isDownloaderImpl(std::lock_guard<std::mutex> & segment_lock) const;
bool isDownloadedUnlocked(std::lock_guard<std::mutex> & segment_lock) const;
void wrapWithCacheInfo(Exception & e, const String & message, std::lock_guard<std::mutex> & segment_lock) const;
bool lastFileSegmentHolder() const;
@ -236,7 +238,8 @@ private:
/// In general case, all file segments are owned by cache.
bool is_detached = false;
std::atomic<bool> is_downloaded{false};
bool is_downloaded{false};
std::atomic<size_t> hits_count = 0; /// cache hits.
std::atomic<size_t> ref_count = 0; /// Used for getting snapshot state

17
src/Common/OvercommitTracker.cpp
View File

@ -14,10 +14,10 @@ using namespace std::chrono_literals;
constexpr std::chrono::microseconds ZERO_MICROSEC = 0us;
OvercommitTracker::OvercommitTracker(std::mutex & global_mutex_)
OvercommitTracker::OvercommitTracker(DB::ProcessList * process_list_)
: picked_tracker(nullptr)
, process_list(process_list_)
, cancellation_state(QueryCancellationState::NONE)
, global_mutex(global_mutex_)
, freed_memory(0)
, required_memory(0)
, next_id(0)
@ -33,11 +33,11 @@ OvercommitResult OvercommitTracker::needToStopQuery(MemoryTracker * tracker, Int
return OvercommitResult::NONE;
// NOTE: Do not change the order of locks
//
// global_mutex must be acquired before overcommit_m, because
// global mutex must be acquired before overcommit_m, because
// method OvercommitTracker::onQueryStop(MemoryTracker *) is
// always called with already acquired global_mutex in
// always called with already acquired global mutex in
// ProcessListEntry::~ProcessListEntry().
std::unique_lock<std::mutex> global_lock(global_mutex);
auto global_lock = process_list->unsafeLock();
std::unique_lock<std::mutex> lk(overcommit_m);
size_t id = next_id++;
@ -137,8 +137,8 @@ void OvercommitTracker::releaseThreads()
cv.notify_all();
}
UserOvercommitTracker::UserOvercommitTracker(DB::ProcessList * process_list, DB::ProcessListForUser * user_process_list_)
: OvercommitTracker(process_list->mutex)
UserOvercommitTracker::UserOvercommitTracker(DB::ProcessList * process_list_, DB::ProcessListForUser * user_process_list_)
: OvercommitTracker(process_list_)
, user_process_list(user_process_list_)
{}
@ -169,8 +169,7 @@ void UserOvercommitTracker::pickQueryToExcludeImpl()
}
GlobalOvercommitTracker::GlobalOvercommitTracker(DB::ProcessList * process_list_)
: OvercommitTracker(process_list_->mutex)
, process_list(process_list_)
: OvercommitTracker(process_list_)
{}
void GlobalOvercommitTracker::pickQueryToExcludeImpl()

31
src/Common/OvercommitTracker.h
View File

@ -36,6 +36,12 @@ struct OvercommitRatio
class MemoryTracker;
namespace DB
{
class ProcessList;
struct ProcessListForUser;
}
enum class OvercommitResult
{
NONE,
@ -71,7 +77,7 @@ struct OvercommitTracker : boost::noncopyable
virtual ~OvercommitTracker() = default;
protected:
explicit OvercommitTracker(std::mutex & global_mutex_);
explicit OvercommitTracker(DB::ProcessList * process_list_);
virtual void pickQueryToExcludeImpl() = 0;
@ -86,6 +92,12 @@ protected:
// overcommit tracker is in SELECTED state.
MemoryTracker * picked_tracker;
// Global mutex stored in ProcessList is used to synchronize
// insertion and deletion of queries.
// OvercommitTracker::pickQueryToExcludeImpl() implementations
// require this mutex to be locked, because they read list (or sublist)
// of queries.
DB::ProcessList * process_list;
private:
void pickQueryToExclude()
@ -113,12 +125,6 @@ private:
QueryCancellationState cancellation_state;
// Global mutex which is used in ProcessList to synchronize
// insertion and deletion of queries.
// OvercommitTracker::pickQueryToExcludeImpl() implementations
// require this mutex to be locked, because they read list (or sublist)
// of queries.
std::mutex & global_mutex;
Int64 freed_memory;
Int64 required_memory;
@ -128,15 +134,9 @@ private:
bool allow_release;
};
namespace DB
{
class ProcessList;
struct ProcessListForUser;
}
struct UserOvercommitTracker : OvercommitTracker
{
explicit UserOvercommitTracker(DB::ProcessList * process_list, DB::ProcessListForUser * user_process_list_);
explicit UserOvercommitTracker(DB::ProcessList * process_list_, DB::ProcessListForUser * user_process_list_);
~UserOvercommitTracker() override = default;
@ -155,9 +155,6 @@ struct GlobalOvercommitTracker : OvercommitTracker
protected:
void pickQueryToExcludeImpl() override;
private:
DB::ProcessList * process_list;
};
// This class is used to disallow tracking during logging to avoid deadlocks.

10
src/Common/mysqlxx/Pool.cpp
View File

@ -9,6 +9,7 @@
#include <mysqlxx/Pool.h>
#include <base/sleep.h>
#include <Poco/Util/LayeredConfiguration.h>
#include <Common/logger_useful.h>
#include <ctime>
@ -260,7 +261,10 @@ void Pool::Entry::forceConnected() const
else
sleepForSeconds(MYSQLXX_POOL_SLEEP_ON_CONNECT_FAIL);
pool->logger.debug("Entry: Reconnecting to MySQL server %s", pool->description);
pool->logger.debug(
"Creating a new MySQL connection to %s with settings: connect_timeout=%u, read_write_timeout=%u",
pool->description, pool->connect_timeout, pool->rw_timeout);
data->conn.connect(
pool->db.c_str(),
pool->server.c_str(),
@ -325,6 +329,10 @@ Pool::Connection * Pool::allocConnection(bool dont_throw_if_failed_first_time)
{
logger.debug("Connecting to %s", description);
logger.debug(
"Creating a new MySQL connection to %s with settings: connect_timeout=%u, read_write_timeout=%u",
description, connect_timeout, rw_timeout);
conn_ptr->conn.connect(
db.c_str(),
server.c_str(),

12
src/Common/mysqlxx/PoolWithFailover.cpp
View File

@ -168,6 +168,7 @@ PoolWithFailover::Entry PoolWithFailover::get()
}
app.logger().warning("Connection to " + pool->getDescription() + " failed: " + e.displayText());
replica_name_to_error_detail.insert_or_assign(pool->getDescription(), ErrorDetail{e.code(), e.displayText()});
continue;
@ -177,7 +178,10 @@ PoolWithFailover::Entry PoolWithFailover::get()
}
}
app.logger().error("Connection to all replicas failed " + std::to_string(try_no + 1) + " times");
if (replicas_by_priority.size() > 1)
app.logger().error("Connection to all mysql replicas failed " + std::to_string(try_no + 1) + " times");
else
app.logger().error("Connection to mysql failed " + std::to_string(try_no + 1) + " times");
}
if (full_pool)
@ -187,7 +191,11 @@ PoolWithFailover::Entry PoolWithFailover::get()
}
DB::WriteBufferFromOwnString message;
message << "Connections to all replicas failed: ";
if (replicas_by_priority.size() > 1)
message << "Connections to all mysql replicas failed: ";
else
message << "Connections to mysql failed: ";
for (auto it = replicas_by_priority.begin(); it != replicas_by_priority.end(); ++it)
{
for (auto jt = it->second.begin(); jt != it->second.end(); ++jt)

22
src/Common/mysqlxx/mysqlxx/Pool.h
View File

@ -169,10 +169,24 @@ public:
unsigned max_connections_ = MYSQLXX_POOL_DEFAULT_MAX_CONNECTIONS,
unsigned enable_local_infile_ = MYSQLXX_DEFAULT_ENABLE_LOCAL_INFILE,
bool opt_reconnect_ = MYSQLXX_DEFAULT_MYSQL_OPT_RECONNECT)
: logger(Poco::Logger::get("mysqlxx::Pool")), default_connections(default_connections_),
max_connections(max_connections_), db(db_), server(server_), user(user_), password(password_), port(port_), socket(socket_),
connect_timeout(connect_timeout_), rw_timeout(rw_timeout_), enable_local_infile(enable_local_infile_),
opt_reconnect(opt_reconnect_) {}
: logger(Poco::Logger::get("mysqlxx::Pool"))
, default_connections(default_connections_)
, max_connections(max_connections_)
, db(db_)
, server(server_)
, user(user_)
, password(password_)
, port(port_)
, socket(socket_)
, connect_timeout(connect_timeout_)
, rw_timeout(rw_timeout_)
, enable_local_infile(enable_local_infile_)
, opt_reconnect(opt_reconnect_)
{
logger.debug(
"Created MySQL Pool with settings: connect_timeout=%u, read_write_timeout=%u, default_connections_number=%u, max_connections_number=%u",
connect_timeout, rw_timeout, default_connections, max_connections);
}
Pool(const Pool & other)
: logger(other.logger), default_connections{other.default_connections},

2
src/Core/Settings.h
View File

@ -629,8 +629,6 @@ static constexpr UInt64 operator""_GiB(unsigned long long value)
M(Bool, allow_experimental_hash_functions, false, "Enable experimental hash functions (hashid, etc)", 0) \
M(Bool, allow_experimental_object_type, false, "Allow Object and JSON data types", 0) \
M(String, insert_deduplication_token, "", "If not empty, used for duplicate detection instead of data digest", 0) \
M(String, ann_index_select_query_params, "", "Parameters passed to ANN indexes in SELECT queries, the format is 'param1=x, param2=y, ...'", 0) \
M(UInt64, max_limit_for_ann_queries, 1000000, "Maximum limit value for using ANN indexes is used to prevent memory overflow in search queries for indexes", 0) \
M(Bool, count_distinct_optimization, false, "Rewrite count distinct to subquery of group by", 0) \
M(Bool, throw_on_unsupported_query_inside_transaction, true, "Throw exception if unsupported query is used inside transaction", 0) \
M(TransactionsWaitCSNMode, wait_changes_become_visible_after_commit_mode, TransactionsWaitCSNMode::WAIT_UNKNOWN, "Wait for committed changes to become actually visible in the latest snapshot", 0) \

15
src/Databases/DatabaseFactory.cpp
View File

@ -183,15 +183,15 @@ DatabasePtr DatabaseFactory::getImpl(const ASTCreateQuery & create, const String
StorageMySQLConfiguration configuration;
ASTs & arguments = engine->arguments->children;
MySQLSettings mysql_settings;
auto mysql_settings = std::make_unique<ConnectionMySQLSettings>();
if (auto named_collection = getExternalDataSourceConfiguration(arguments, context, true, true, mysql_settings))
if (auto named_collection = getExternalDataSourceConfiguration(arguments, context, true, true, *mysql_settings))
{
auto [common_configuration, storage_specific_args, settings_changes] = named_collection.value();
configuration.set(common_configuration);
configuration.addresses = {std::make_pair(configuration.host, configuration.port)};
mysql_settings.applyChanges(settings_changes);
mysql_settings->applyChanges(settings_changes);
if (!storage_specific_args.empty())
throw Exception(ErrorCodes::BAD_ARGUMENTS,
@ -228,15 +228,14 @@ DatabasePtr DatabaseFactory::getImpl(const ASTCreateQuery & create, const String
{
if (engine_name == "MySQL")
{
auto mysql_database_settings = std::make_unique<ConnectionMySQLSettings>();
auto mysql_pool = createMySQLPoolWithFailover(configuration, mysql_settings);
mysql_settings->loadFromQueryContext(context);
mysql_settings->loadFromQuery(*engine_define); /// higher priority
mysql_database_settings->loadFromQueryContext(context);
mysql_database_settings->loadFromQuery(*engine_define); /// higher priority
auto mysql_pool = createMySQLPoolWithFailover(configuration, *mysql_settings);
return std::make_shared<DatabaseMySQL>(
context, database_name, metadata_path, engine_define, configuration.database,
std::move(mysql_database_settings), std::move(mysql_pool), create.attach);
std::move(mysql_settings), std::move(mysql_pool), create.attach);
}
MySQLClient client(configuration.host, configuration.port, configuration.username, configuration.password);

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

@ -703,22 +703,6 @@ bool CachedOnDiskReadBufferFromFile::updateImplementationBufferIfNeeded()
size_t download_offset = file_segment->getDownloadOffset();
bool cached_part_is_finished = download_offset == file_offset_of_buffer_end;
#ifndef NDEBUG
size_t cache_file_size = getFileSizeFromReadBuffer(*implementation_buffer);
size_t cache_file_read_offset = implementation_buffer->getFileOffsetOfBufferEnd();
size_t implementation_buffer_finished = cache_file_size == cache_file_read_offset;
if (cached_part_is_finished != implementation_buffer_finished)
{
throw Exception(
ErrorCodes::LOGICAL_ERROR,
"Incorrect state of buffers. Current download offset: {}, file offset of buffer end: {}, "
"cache file size: {}, cache file offset: {}, file segment info: {}",
download_offset, file_offset_of_buffer_end, cache_file_size, cache_file_read_offset,
file_segment->getInfoForLog());
}
#endif
if (cached_part_is_finished)
{
/// TODO: makes sense to reuse local file reader if we return here with CACHED read type again?

12
src/Interpreters/ProcessList.cpp
View File

@ -82,7 +82,7 @@ ProcessList::EntryPtr ProcessList::insert(const String & query_, const IAST * as
bool is_unlimited_query = isUnlimitedQuery(ast);
{
std::unique_lock lock(mutex);
auto [lock, overcommit_blocker] = safeLock(); // To avoid deadlock in case of OOM
IAST::QueryKind query_kind = ast->getQueryKind();
const auto queue_max_wait_ms = settings.queue_max_wait_ms.totalMilliseconds();
@ -269,7 +269,7 @@ ProcessList::EntryPtr ProcessList::insert(const String & query_, const IAST * as
ProcessListEntry::~ProcessListEntry()
{
std::lock_guard lock(parent.mutex);
auto lock = parent.safeLock();
String user = it->getClientInfo().current_user;
String query_id = it->getClientInfo().current_query_id;
@ -430,7 +430,7 @@ QueryStatus * ProcessList::tryGetProcessListElement(const String & current_query
CancellationCode ProcessList::sendCancelToQuery(const String & current_query_id, const String & current_user, bool kill)
{
std::lock_guard lock(mutex);
auto lock = safeLock();
QueryStatus * elem = tryGetProcessListElement(current_query_id, current_user);
@ -443,7 +443,7 @@ CancellationCode ProcessList::sendCancelToQuery(const String & current_query_id,
void ProcessList::killAllQueries()
{
std::lock_guard lock(mutex);
auto lock = safeLock();
for (auto & process : processes)
process.cancelQuery(true);
@ -495,7 +495,7 @@ ProcessList::Info ProcessList::getInfo(bool get_thread_list, bool get_profile_ev
{
Info per_query_infos;
std::lock_guard lock(mutex);
auto lock = safeLock();
per_query_infos.reserve(processes.size());
for (const auto & process : processes)
@ -530,7 +530,7 @@ ProcessList::UserInfo ProcessList::getUserInfo(bool get_profile_events) const
{
UserInfo per_user_infos;
std::lock_guard lock(mutex);
auto lock = safeLock();
per_user_infos.reserve(user_to_queries.size());

29
src/Interpreters/ProcessList.h
View File

@ -285,7 +285,26 @@ public:
};
class ProcessList
class ProcessListBase
{
mutable std::mutex mutex;
protected:
using Lock = std::unique_lock<std::mutex>;
struct LockAndBlocker
{
Lock lock;
OvercommitTrackerBlockerInThread blocker;
};
// It is forbidden to do allocations/deallocations with acquired mutex and
// enabled OvercommitTracker. This leads to deadlock in the case of OOM.
LockAndBlocker safeLock() const noexcept { return { std::unique_lock{mutex}, {} }; }
Lock unsafeLock() const noexcept { return std::unique_lock{mutex}; }
};
class ProcessList : public ProcessListBase
{
public:
using Element = QueryStatus;
@ -304,10 +323,10 @@ public:
protected:
friend class ProcessListEntry;
friend struct ::OvercommitTracker;
friend struct ::UserOvercommitTracker;
friend struct ::GlobalOvercommitTracker;
mutable std::mutex mutex;
mutable std::condition_variable have_space; /// Number of currently running queries has become less than maximum.
/// List of queries
@ -360,19 +379,19 @@ public:
void setMaxSize(size_t max_size_)
{
std::lock_guard lock(mutex);
auto lock = unsafeLock();
max_size = max_size_;
}
void setMaxInsertQueriesAmount(size_t max_insert_queries_amount_)
{
std::lock_guard lock(mutex);
auto lock = unsafeLock();
max_insert_queries_amount = max_insert_queries_amount_;
}
void setMaxSelectQueriesAmount(size_t max_select_queries_amount_)
{
std::lock_guard lock(mutex);
auto lock = unsafeLock();
max_select_queries_amount = max_select_queries_amount_;
}

11
src/Interpreters/RequiredSourceColumnsVisitor.cpp
View File

@ -154,17 +154,6 @@ void RequiredSourceColumnsMatcher::visit(const ASTSelectQuery & select, const AS
find_columns(interpolate->as<ASTInterpolateElement>()->expr.get());
}
if (const auto & with = select.with())
{
for (auto & node : with->children)
{
if (const auto * identifier = node->as<ASTIdentifier>())
data.addColumnIdentifier(*identifier);
else
data.addColumnAliasIfAny(*node);
}
}
std::vector<ASTPtr *> out;
for (const auto & node : select.children)
{

5
src/Server/GRPCServer.cpp
View File

@ -1862,6 +1862,11 @@ void GRPCServer::start()
queue = builder.AddCompletionQueue();
grpc_server = builder.BuildAndStart();
if (nullptr == grpc_server)
{
throw DB::Exception("Can't start grpc server, there is a port conflict", DB::ErrorCodes::NETWORK_ERROR);
}
runner->start();
}

6
src/Storages/ExternalDataSourceConfiguration.cpp
View File

@ -17,6 +17,7 @@
#endif
#if USE_MYSQL
#include <Storages/MySQL/MySQLSettings.h>
#include <Databases/MySQL/ConnectionMySQLSettings.h>
#endif
#if USE_NATSIO
#include <Storages/NATS/NATSSettings.h>
@ -575,6 +576,10 @@ template
std::optional<ExternalDataSourceInfo> getExternalDataSourceConfiguration(
const ASTs & args, ContextPtr context, bool is_database_engine, bool throw_on_no_collection, const BaseSettings<MySQLSettingsTraits> & storage_settings);
template
std::optional<ExternalDataSourceInfo> getExternalDataSourceConfiguration(
const ASTs & args, ContextPtr context, bool is_database_engine, bool throw_on_no_collection, const BaseSettings<ConnectionMySQLSettingsTraits> & storage_settings);
template
std::optional<ExternalDataSourceInfo> getExternalDataSourceConfiguration(
const Poco::Util::AbstractConfiguration & dict_config, const String & dict_config_prefix,
@ -583,5 +588,6 @@ std::optional<ExternalDataSourceInfo> getExternalDataSourceConfiguration(
template
SettingsChanges getSettingsChangesFromConfig(
const BaseSettings<MySQLSettingsTraits> & settings, const Poco::Util::AbstractConfiguration & config, const String & config_prefix);
#endif
}

595
src/Storages/MergeTree/CommonANNIndexes.cpp
View File

@ -1,595 +0,0 @@
#include <Storages/MergeTree/CommonANNIndexes.h>
#include <Storages/MergeTree/KeyCondition.h>
#include <Parsers/ASTFunction.h>
#include <Parsers/ASTIdentifier.h>
#include <Parsers/ASTLiteral.h>
#include <Parsers/ASTOrderByElement.h>
#include <Parsers/ASTSelectQuery.h>
#include <Parsers/ASTSetQuery.h>
#include <Storages/MergeTree/MergeTreeSettings.h>
#include <Interpreters/Context.h>
namespace DB
{
namespace ErrorCodes
{
extern const int LOGICAL_ERROR;
extern const int INCORRECT_QUERY;
}
namespace
{
namespace ANN = ApproximateNearestNeighbour;
template <typename Literal>
void extractTargetVectorFromLiteral(ANN::ANNQueryInformation::Embedding & target, Literal literal)
{
Float64 float_element_of_target_vector;
Int64 int_element_of_target_vector;
for (const auto & value : literal.value())
{
if (value.tryGet(float_element_of_target_vector))
{
target.emplace_back(float_element_of_target_vector);
}
else if (value.tryGet(int_element_of_target_vector))
{
target.emplace_back(static_cast<float>(int_element_of_target_vector));
}
else
{
throw Exception(ErrorCodes::INCORRECT_QUERY, "Wrong type of elements in target vector. Only float or int are supported.");
}
}
}
ANN::ANNQueryInformation::Metric castMetricFromStringToType(String metric_name)
{
if (metric_name == "L2Distance")
return ANN::ANNQueryInformation::Metric::L2;
if (metric_name == "LpDistance")
return ANN::ANNQueryInformation::Metric::Lp;
return ANN::ANNQueryInformation::Metric::Unknown;
}
}
namespace ApproximateNearestNeighbour
{
ANNCondition::ANNCondition(const SelectQueryInfo & query_info,
ContextPtr context) :
block_with_constants{KeyCondition::getBlockWithConstants(query_info.query, query_info.syntax_analyzer_result, context)},
ann_index_select_query_params{context->getSettings().get("ann_index_select_query_params").get<String>()},
index_granularity{context->getMergeTreeSettings().get("index_granularity").get<UInt64>()},
limit_restriction{context->getSettings().get("max_limit_for_ann_queries").get<UInt64>()},
index_is_useful{checkQueryStructure(query_info)} {}
bool ANNCondition::alwaysUnknownOrTrue(String metric_name) const
{
if (!index_is_useful)
{
return true; // Query isn't supported
}
// If query is supported, check metrics for match
return !(castMetricFromStringToType(metric_name) == query_information->metric);
}
float ANNCondition::getComparisonDistanceForWhereQuery() const
{
if (index_is_useful && query_information.has_value()
&& query_information->query_type == ANNQueryInformation::Type::Where)
{
return query_information->distance;
}
throw Exception(ErrorCodes::LOGICAL_ERROR, "Not supported method for this query type");
}
UInt64 ANNCondition::getLimit() const
{
if (index_is_useful && query_information.has_value())
{
return query_information->limit;
}
throw Exception(ErrorCodes::LOGICAL_ERROR, "No LIMIT section in query, not supported");
}
std::vector<float> ANNCondition::getTargetVector() const
{
if (index_is_useful && query_information.has_value())
{
return query_information->target;
}
throw Exception(ErrorCodes::LOGICAL_ERROR, "Target vector was requested for useless or uninitialized index.");
}
size_t ANNCondition::getNumOfDimensions() const
{
if (index_is_useful && query_information.has_value())
{
return query_information->target.size();
}
throw Exception(ErrorCodes::LOGICAL_ERROR, "Number of dimensions was requested for useless or uninitialized index.");
}
String ANNCondition::getColumnName() const
{
if (index_is_useful && query_information.has_value())
{
return query_information->column_name;
}
throw Exception(ErrorCodes::LOGICAL_ERROR, "Column name was requested for useless or uninitialized index.");
}
ANNQueryInformation::Metric ANNCondition::getMetricType() const
{
if (index_is_useful && query_information.has_value())
{
return query_information->metric;
}
throw Exception(ErrorCodes::LOGICAL_ERROR, "Metric name was requested for useless or uninitialized index.");
}
float ANNCondition::getPValueForLpDistance() const
{
if (index_is_useful && query_information.has_value())
{
return query_information->p_for_lp_dist;
}
throw Exception(ErrorCodes::LOGICAL_ERROR, "P from LPDistance was requested for useless or uninitialized index.");
}
ANNQueryInformation::Type ANNCondition::getQueryType() const
{
if (index_is_useful && query_information.has_value())
{
return query_information->query_type;
}
throw Exception(ErrorCodes::LOGICAL_ERROR, "Query type was requested for useless or uninitialized index.");
}
bool ANNCondition::checkQueryStructure(const SelectQueryInfo & query)
{
// RPN-s for different sections of the query
RPN rpn_prewhere_clause;
RPN rpn_where_clause;
RPN rpn_order_by_clause;
RPNElement rpn_limit;
UInt64 limit;
ANNQueryInformation prewhere_info;
ANNQueryInformation where_info;
ANNQueryInformation order_by_info;
// Build rpns for query sections
const auto & select = query.query->as<ASTSelectQuery &>();
if (select.prewhere()) // If query has PREWHERE clause
{
traverseAST(select.prewhere(), rpn_prewhere_clause);
}
if (select.where()) // If query has WHERE clause
{
traverseAST(select.where(), rpn_where_clause);
}
if (select.limitLength()) // If query has LIMIT clause
{
traverseAtomAST(select.limitLength(), rpn_limit);
}
if (select.orderBy()) // If query has ORDERBY clause
{
traverseOrderByAST(select.orderBy(), rpn_order_by_clause);
}
// Reverse RPNs for conveniences during parsing
std::reverse(rpn_prewhere_clause.begin(), rpn_prewhere_clause.end());
std::reverse(rpn_where_clause.begin(), rpn_where_clause.end());
std::reverse(rpn_order_by_clause.begin(), rpn_order_by_clause.end());
// Match rpns with supported types and extract information
const bool prewhere_is_valid = matchRPNWhere(rpn_prewhere_clause, prewhere_info);
const bool where_is_valid = matchRPNWhere(rpn_where_clause, where_info);
const bool order_by_is_valid = matchRPNOrderBy(rpn_order_by_clause, order_by_info);
const bool limit_is_valid = matchRPNLimit(rpn_limit, limit);
// Query without a LIMIT clause or with a limit greater than a restriction is not supported
if (!limit_is_valid || limit_restriction < limit)
{
return false;
}
// Search type query in both sections isn't supported
if (prewhere_is_valid && where_is_valid)
{
return false;
}
// Search type should be in WHERE or PREWHERE clause
if (prewhere_is_valid || where_is_valid)
{
query_information = std::move(prewhere_is_valid ? prewhere_info : where_info);
}
if (order_by_is_valid)
{
// Query with valid where and order by type is not supported
if (query_information.has_value())
{
return false;
}
query_information = std::move(order_by_info);
}
if (query_information)
query_information->limit = limit;
return query_information.has_value();
}
void ANNCondition::traverseAST(const ASTPtr & node, RPN & rpn)
{
// If the node is ASTFunction, it may have children nodes
if (const auto * func = node->as<ASTFunction>())
{
const ASTs & children = func->arguments->children;
// Traverse children nodes
for (const auto& child : children)
{
traverseAST(child, rpn);
}
}
RPNElement element;
// Get the data behind node
if (!traverseAtomAST(node, element))
{
element.function = RPNElement::FUNCTION_UNKNOWN;
}
rpn.emplace_back(std::move(element));
}
bool ANNCondition::traverseAtomAST(const ASTPtr & node, RPNElement & out)
{
// Match Functions
if (const auto * function = node->as<ASTFunction>())
{
// Set the name
out.func_name = function->name;
if (function->name == "L1Distance" ||
function->name == "L2Distance" ||
function->name == "LinfDistance" ||
function->name == "cosineDistance" ||
function->name == "dotProduct" ||
function->name == "LpDistance")
{
out.function = RPNElement::FUNCTION_DISTANCE;
}
else if (function->name == "tuple")
{
out.function = RPNElement::FUNCTION_TUPLE;
}
else if (function->name == "array")
{
out.function = RPNElement::FUNCTION_ARRAY;
}
else if (function->name == "less" ||
function->name == "greater" ||
function->name == "lessOrEquals" ||
function->name == "greaterOrEquals")
{
out.function = RPNElement::FUNCTION_COMPARISON;
}
else if (function->name == "_CAST")
{
out.function = RPNElement::FUNCTION_CAST;
}
else
{
return false;
}
return true;
}
// Match identifier
else if (const auto * identifier = node->as<ASTIdentifier>())
{
out.function = RPNElement::FUNCTION_IDENTIFIER;
out.identifier.emplace(identifier->name());
out.func_name = "column identifier";
return true;
}
// Check if we have constants behind the node
return tryCastToConstType(node, out);
}
bool ANNCondition::tryCastToConstType(const ASTPtr & node, RPNElement & out)
{
Field const_value;
DataTypePtr const_type;
if (KeyCondition::getConstant(node, block_with_constants, const_value, const_type))
{
/// Check for constant types
if (const_value.getType() == Field::Types::Float64)
{
out.function = RPNElement::FUNCTION_FLOAT_LITERAL;
out.float_literal.emplace(const_value.get<Float32>());
out.func_name = "Float literal";
return true;
}
if (const_value.getType() == Field::Types::UInt64)
{
out.function = RPNElement::FUNCTION_INT_LITERAL;
out.int_literal.emplace(const_value.get<UInt64>());
out.func_name = "Int literal";
return true;
}
if (const_value.getType() == Field::Types::Int64)
{
out.function = RPNElement::FUNCTION_INT_LITERAL;
out.int_literal.emplace(const_value.get<Int64>());
out.func_name = "Int literal";
return true;
}
if (const_value.getType() == Field::Types::Tuple)
{
out.function = RPNElement::FUNCTION_LITERAL_TUPLE;
out.tuple_literal = const_value.get<Tuple>();
out.func_name = "Tuple literal";
return true;
}
if (const_value.getType() == Field::Types::Array)
{
out.function = RPNElement::FUNCTION_LITERAL_ARRAY;
out.array_literal = const_value.get<Array>();
out.func_name = "Array literal";
return true;
}
if (const_value.getType() == Field::Types::String)
{
out.function = RPNElement::FUNCTION_STRING_LITERAL;
out.func_name = const_value.get<String>();
return true;
}
}
return false;
}
void ANNCondition::traverseOrderByAST(const ASTPtr & node, RPN & rpn)
{
if (const auto * expr_list = node->as<ASTExpressionList>())
{
if (const auto * order_by_element = expr_list->children.front()->as<ASTOrderByElement>())
{
traverseAST(order_by_element->children.front(), rpn);
}
}
}
// Returns true and stores ANNQueryInformation if the query has valid WHERE clause
bool ANNCondition::matchRPNWhere(RPN & rpn, ANNQueryInformation & expr)
{
// WHERE section must have at least 5 expressions
// Operator->Distance(float)->DistanceFunc->Column->Tuple(Array)Func(TargetVector(floats))
if (rpn.size() < 5)
{
return false;
}
auto iter = rpn.begin();
// Query starts from operator less
if (iter->function != RPNElement::FUNCTION_COMPARISON)
{
return false;
}
const bool greater_case = iter->func_name == "greater" || iter->func_name == "greaterOrEquals";
const bool less_case = iter->func_name == "less" || iter->func_name == "lessOrEquals";
++iter;
if (less_case)
{
if (iter->function != RPNElement::FUNCTION_FLOAT_LITERAL)
{
return false;
}
expr.distance = getFloatOrIntLiteralOrPanic(iter);
++iter;
}
else if (!greater_case)
{
return false;
}
auto end = rpn.end();
if (!matchMainParts(iter, end, expr))
{
return false;
}
if (greater_case)
{
if (expr.target.size() < 2)
{
return false;
}
expr.distance = expr.target.back();
expr.target.pop_back();
}
// query is ok
return true;
}
// Returns true and stores ANNExpr if the query has valid ORDERBY clause
bool ANNCondition::matchRPNOrderBy(RPN & rpn, ANNQueryInformation & expr)
{
// ORDER BY clause must have at least 3 expressions
if (rpn.size() < 3)
{
return false;
}
auto iter = rpn.begin();
auto end = rpn.end();
return ANNCondition::matchMainParts(iter, end, expr);
}
// Returns true and stores Length if we have valid LIMIT clause in query
bool ANNCondition::matchRPNLimit(RPNElement & rpn, UInt64 & limit)
{
if (rpn.function == RPNElement::FUNCTION_INT_LITERAL)
{
limit = rpn.int_literal.value();
return true;
}
return false;
}
/* Matches dist function, target vector, column name */
bool ANNCondition::matchMainParts(RPN::iterator & iter, const RPN::iterator & end, ANNQueryInformation & expr)
{
bool identifier_found = false;
// Matches DistanceFunc->[Column]->[Tuple(array)Func]->TargetVector(floats)->[Column]
if (iter->function != RPNElement::FUNCTION_DISTANCE)
{
return false;
}
expr.metric = castMetricFromStringToType(iter->func_name);
++iter;
if (expr.metric == ANN::ANNQueryInformation::Metric::Lp)
{
if (iter->function != RPNElement::FUNCTION_FLOAT_LITERAL &&
iter->function != RPNElement::FUNCTION_INT_LITERAL)
{
return false;
}
expr.p_for_lp_dist = getFloatOrIntLiteralOrPanic(iter);
++iter;
}
if (iter->function == RPNElement::FUNCTION_IDENTIFIER)
{
identifier_found = true;
expr.column_name = std::move(iter->identifier.value());
++iter;
}
if (iter->function == RPNElement::FUNCTION_TUPLE || iter->function == RPNElement::FUNCTION_ARRAY)
{
++iter;
}
if (iter->function == RPNElement::FUNCTION_LITERAL_TUPLE)
{
extractTargetVectorFromLiteral(expr.target, iter->tuple_literal);
++iter;
}
if (iter->function == RPNElement::FUNCTION_LITERAL_ARRAY)
{
extractTargetVectorFromLiteral(expr.target, iter->array_literal);
++iter;
}
/// further conditions are possible if there is no tuple or array, or no identifier is found
/// the tuple or array can be inside a cast function. For other cases, see the loop after this condition
if (iter != end && iter->function == RPNElement::FUNCTION_CAST)
{
++iter;
/// Cast should be made to array or tuple
if (!iter->func_name.starts_with("Array") && !iter->func_name.starts_with("Tuple"))
{
return false;
}
++iter;
if (iter->function == RPNElement::FUNCTION_LITERAL_TUPLE)
{
extractTargetVectorFromLiteral(expr.target, iter->tuple_literal);
++iter;
}
else if (iter->function == RPNElement::FUNCTION_LITERAL_ARRAY)
{
extractTargetVectorFromLiteral(expr.target, iter->array_literal);
++iter;
}
else
{
return false;
}
}
while (iter != end)
{
if (iter->function == RPNElement::FUNCTION_FLOAT_LITERAL ||
iter->function == RPNElement::FUNCTION_INT_LITERAL)
{
expr.target.emplace_back(getFloatOrIntLiteralOrPanic(iter));
}
else if (iter->function == RPNElement::FUNCTION_IDENTIFIER)
{
if (identifier_found)
{
return false;
}
expr.column_name = std::move(iter->identifier.value());
identifier_found = true;
}
else
{
return false;
}
++iter;
}
// Final checks of correctness
return identifier_found && !expr.target.empty();
}
// Gets float or int from AST node
float ANNCondition::getFloatOrIntLiteralOrPanic(const RPN::iterator& iter)
{
if (iter->float_literal.has_value())
{
return iter->float_literal.value();
}
if (iter->int_literal.has_value())
{
return static_cast<float>(iter->int_literal.value());
}
throw Exception("Wrong parsed AST in buildRPN\n", ErrorCodes::INCORRECT_QUERY);
}
}
}

236
src/Storages/MergeTree/CommonANNIndexes.h
View File

@ -1,236 +0,0 @@
#pragma once
#include <Storages/MergeTree/MergeTreeIndices.h>
#include "base/types.h"
#include <optional>
#include <vector>
namespace DB
{
namespace ApproximateNearestNeighbour
{
/**
* Queries for Approximate Nearest Neighbour Search
* have similar structure:
* 1) target vector from which all distances are calculated
* 2) metric name (e.g L2Distance, LpDistance, etc.)
* 3) name of column with embeddings
* 4) type of query
* 5) Number of elements, that should be taken (limit)
*
* And two optional parameters:
* 1) p for LpDistance function
* 2) distance to compare with (only for where queries)
*/
struct ANNQueryInformation
{
using Embedding = std::vector<float>;
// Extracted data from valid query
Embedding target;
enum class Metric
{
Unknown,
L2,
Lp
} metric;
String column_name;
UInt64 limit;
enum class Type
{
OrderBy,
Where
} query_type;
float p_for_lp_dist = -1.0;
float distance = -1.0;
};
/**
Class ANNCondition, is responsible for recognizing special query types which
can be speeded up by ANN Indexes. It parses the SQL query and checks
if it matches ANNIndexes. The recognizing method - alwaysUnknownOrTrue
returns false if we can speed up the query, and true otherwise.
It has only one argument, name of the metric with which index was built.
There are two main patterns of queries being supported
1) Search query type
SELECT * FROM * WHERE DistanceFunc(column, target_vector) < floatLiteral LIMIT count
2) OrderBy query type
SELECT * FROM * WHERE * ORDERBY DistanceFunc(column, target_vector) LIMIT count
*Query without LIMIT count is not supported*
target_vector(should have float coordinates) examples:
tuple(0.1, 0.1, ...., 0.1) or (0.1, 0.1, ...., 0.1)
[the word tuple is not needed]
If the query matches one of these two types, than the class extracts useful information
from the query. If the query has both 1 and 2 types, than we can't speed and alwaysUnknownOrTrue
returns true.
From matching query it extracts
* targetVector
* metricName(DistanceFunction)
* dimension size if query uses LpDistance
* distance to compare(ONLY for search types, otherwise you get exception)
* spaceDimension(which is targetVector's components count)
* column
* objects count from LIMIT clause(for both queries)
* settings str, if query has settings section with new 'ann_index_select_query_params' value,
than you can get the new value(empty by default) calling method getSettingsStr
* queryHasOrderByClause and queryHasWhereClause return true if query matches the type
Search query type is also recognized for PREWHERE clause
*/
class ANNCondition
{
public:
ANNCondition(const SelectQueryInfo & query_info,
ContextPtr context);
// false if query can be speeded up, true otherwise
bool alwaysUnknownOrTrue(String metric_name) const;
// returns the distance to compare with for search query
float getComparisonDistanceForWhereQuery() const;
// distance should be calculated regarding to targetVector
std::vector<float> getTargetVector() const;
// targetVector dimension size
size_t getNumOfDimensions() const;
String getColumnName() const;
ANNQueryInformation::Metric getMetricType() const;
// the P- value if the metric is 'LpDistance'
float getPValueForLpDistance() const;
ANNQueryInformation::Type getQueryType() const;
UInt64 getIndexGranularity() const { return index_granularity; }
// length's value from LIMIT clause
UInt64 getLimit() const;
// value of 'ann_index_select_query_params' if have in SETTINGS clause, empty string otherwise
String getParamsStr() const { return ann_index_select_query_params; }
private:
struct RPNElement
{
enum Function
{
// DistanceFunctions
FUNCTION_DISTANCE,
//tuple(0.1, ..., 0.1)
FUNCTION_TUPLE,
//array(0.1, ..., 0.1)
FUNCTION_ARRAY,
// Operators <, >, <=, >=
FUNCTION_COMPARISON,
// Numeric float value
FUNCTION_FLOAT_LITERAL,
// Numeric int value
FUNCTION_INT_LITERAL,
// Column identifier
FUNCTION_IDENTIFIER,
// Unknown, can be any value
FUNCTION_UNKNOWN,
// (0.1, ...., 0.1) vector without word 'tuple'
FUNCTION_LITERAL_TUPLE,
// [0.1, ...., 0.1] vector without word 'array'
FUNCTION_LITERAL_ARRAY,
// if client parameters are used, cast will always be in the query
FUNCTION_CAST,
// name of type in cast function
FUNCTION_STRING_LITERAL,
};
explicit RPNElement(Function function_ = FUNCTION_UNKNOWN)
: function(function_), func_name("Unknown"), float_literal(std::nullopt), identifier(std::nullopt) {}
Function function;
String func_name;
std::optional<float> float_literal;
std::optional<String> identifier;
std::optional<int64_t> int_literal;
std::optional<Tuple> tuple_literal;
std::optional<Array> array_literal;
UInt32 dim = 0;
};
using RPN = std::vector<RPNElement>;
bool checkQueryStructure(const SelectQueryInfo & query);
// Util functions for the traversal of AST, parses AST and builds rpn
void traverseAST(const ASTPtr & node, RPN & rpn);
// Return true if we can identify our node type
bool traverseAtomAST(const ASTPtr & node, RPNElement & out);
// Checks if the AST stores ConstType expression
bool tryCastToConstType(const ASTPtr & node, RPNElement & out);
// Traverses the AST of ORDERBY section
void traverseOrderByAST(const ASTPtr & node, RPN & rpn);
// Returns true and stores ANNExpr if the query has valid WHERE section
static bool matchRPNWhere(RPN & rpn, ANNQueryInformation & expr);
// Returns true and stores ANNExpr if the query has valid ORDERBY section
static bool matchRPNOrderBy(RPN & rpn, ANNQueryInformation & expr);
// Returns true and stores Length if we have valid LIMIT clause in query
static bool matchRPNLimit(RPNElement & rpn, UInt64 & limit);
/* Matches dist function, target vector, column name */
static bool matchMainParts(RPN::iterator & iter, const RPN::iterator & end, ANNQueryInformation & expr);
// Gets float or int from AST node
static float getFloatOrIntLiteralOrPanic(const RPN::iterator& iter);
Block block_with_constants;
// true if we have one of two supported query types
std::optional<ANNQueryInformation> query_information;
// Get from settings ANNIndex parameters
String ann_index_select_query_params;
UInt64 index_granularity;
/// only queries with a lower limit can be considered to avoid memory overflow
UInt64 limit_restriction;
bool index_is_useful = false;
};
// condition interface for Ann indexes. Returns vector of indexes of ranges in granule which are useful for query.
class IMergeTreeIndexConditionAnn : public IMergeTreeIndexCondition
{
public:
virtual std::vector<size_t> getUsefulRanges(MergeTreeIndexGranulePtr idx_granule) const = 0;
};
}
}

27
src/Storages/MergeTree/MergeTreeDataSelectExecutor.cpp
View File

@ -43,8 +43,6 @@
#include <IO/WriteBufferFromOStream.h>
#include <Storages/MergeTree/CommonANNIndexes.h>
namespace DB
{
@ -1671,31 +1669,6 @@ MarkRanges MergeTreeDataSelectExecutor::filterMarksUsingIndex(
{
if (index_mark != index_range.begin || !granule || last_index_mark != index_range.begin)
granule = reader.read();
// Cast to Ann condition
auto ann_condition = std::dynamic_pointer_cast<ApproximateNearestNeighbour::IMergeTreeIndexConditionAnn>(condition);
if (ann_condition != nullptr)
{
// vector of indexes of useful ranges
auto result = ann_condition->getUsefulRanges(granule);
if (result.empty())
{
++granules_dropped;
}
for (auto range : result)
{
// range for corresponding index
MarkRange data_range(
std::max(ranges[i].begin, index_mark * index_granularity + range),
std::min(ranges[i].end, index_mark * index_granularity + range + 1));
if (res.empty() || res.back().end - data_range.begin > min_marks_for_seek)
res.push_back(data_range);
else
res.back().end = data_range.end;
}
continue;
}
if (!condition->mayBeTrueOnGranule(granule))
{

317
src/Storages/MergeTree/MergeTreeIndexAnnoy.cpp
View File

@ -1,317 +0,0 @@
#ifdef ENABLE_ANNOY
#include <Storages/MergeTree/MergeTreeIndexAnnoy.h>
#include <Common/typeid_cast.h>
#include <Core/Field.h>
#include <IO/ReadHelpers.h>
#include <IO/WriteHelpers.h>
#include <Interpreters/castColumn.h>
#include <Columns/ColumnArray.h>
#include <DataTypes/DataTypeArray.h>
namespace DB
{
namespace ApproximateNearestNeighbour
{
template<typename Dist>
void AnnoyIndex<Dist>::serialize(WriteBuffer& ostr) const
{
assert(Base::_built);
writeIntBinary(Base::_s, ostr);
writeIntBinary(Base::_n_items, ostr);
writeIntBinary(Base::_n_nodes, ostr);
writeIntBinary(Base::_nodes_size, ostr);
writeIntBinary(Base::_K, ostr);
writeIntBinary(Base::_seed, ostr);
writeVectorBinary(Base::_roots, ostr);
ostr.write(reinterpret_cast<const char*>(Base::_nodes), Base::_s * Base::_n_nodes);
}
template<typename Dist>
void AnnoyIndex<Dist>::deserialize(ReadBuffer& istr)
{
assert(!Base::_built);
readIntBinary(Base::_s, istr);
readIntBinary(Base::_n_items, istr);
readIntBinary(Base::_n_nodes, istr);
readIntBinary(Base::_nodes_size, istr);
readIntBinary(Base::_K, istr);
readIntBinary(Base::_seed, istr);
readVectorBinary(Base::_roots, istr);
Base::_nodes = realloc(Base::_nodes, Base::_s * Base::_n_nodes);
istr.read(reinterpret_cast<char*>(Base::_nodes), Base::_s * Base::_n_nodes);
Base::_fd = 0;
// set flags
Base::_loaded = false;
Base::_verbose = false;
Base::_on_disk = false;
Base::_built = true;
}
template<typename Dist>
uint64_t AnnoyIndex<Dist>::getNumOfDimensions() const
{
return Base::get_f();
}
}
namespace ErrorCodes
{
extern const int LOGICAL_ERROR;
extern const int INCORRECT_QUERY;
extern const int INCORRECT_DATA;
}
MergeTreeIndexGranuleAnnoy::MergeTreeIndexGranuleAnnoy(const String & index_name_, const Block & index_sample_block_)
: index_name(index_name_)
, index_sample_block(index_sample_block_)
, index(nullptr)
{}
MergeTreeIndexGranuleAnnoy::MergeTreeIndexGranuleAnnoy(
const String & index_name_,
const Block & index_sample_block_,
AnnoyIndexPtr index_base_)
: index_name(index_name_)
, index_sample_block(index_sample_block_)
, index(std::move(index_base_))
{}
void MergeTreeIndexGranuleAnnoy::serializeBinary(WriteBuffer & ostr) const
{
/// number of dimensions is required in the constructor,
/// so it must be written and read separately from the other part
writeIntBinary(index->getNumOfDimensions(), ostr); // write dimension
index->serialize(ostr);
}
void MergeTreeIndexGranuleAnnoy::deserializeBinary(ReadBuffer & istr, MergeTreeIndexVersion /*version*/)
{
uint64_t dimension;
readIntBinary(dimension, istr);
index = std::make_shared<AnnoyIndex>(dimension);
index->deserialize(istr);
}
MergeTreeIndexAggregatorAnnoy::MergeTreeIndexAggregatorAnnoy(
const String & index_name_,
const Block & index_sample_block_,
uint64_t number_of_trees_)
: index_name(index_name_)
, index_sample_block(index_sample_block_)
, number_of_trees(number_of_trees_)
{}
MergeTreeIndexGranulePtr MergeTreeIndexAggregatorAnnoy::getGranuleAndReset()
{
// NOLINTNEXTLINE(*)
index->build(number_of_trees, /*number_of_threads=*/1);
auto granule = std::make_shared<MergeTreeIndexGranuleAnnoy>(index_name, index_sample_block, index);
index = nullptr;
return granule;
}
void MergeTreeIndexAggregatorAnnoy::update(const Block & block, size_t * pos, size_t limit)
{
if (*pos >= block.rows())
throw Exception(
ErrorCodes::LOGICAL_ERROR,
"The provided position is not less than the number of block rows. Position: {}, Block rows: {}.",
toString(*pos), toString(block.rows()));
size_t rows_read = std::min(limit, block.rows() - *pos);
if (index_sample_block.columns() > 1)
{
throw Exception("Only one column is supported", ErrorCodes::LOGICAL_ERROR);
}
auto index_column_name = index_sample_block.getByPosition(0).name;
const auto & column_cut = block.getByName(index_column_name).column->cut(*pos, rows_read);
const auto & column_array = typeid_cast<const ColumnArray*>(column_cut.get());
if (column_array)
{
const auto & data = column_array->getData();
const auto & array = typeid_cast<const ColumnFloat32&>(data).getData();
const auto & offsets = column_array->getOffsets();
size_t num_rows = column_array->size();
/// All sizes are the same
size_t size = offsets[0];
for (size_t i = 0; i < num_rows - 1; ++i)
{
if (offsets[i + 1] - offsets[i] != size)
{
throw Exception(ErrorCodes::INCORRECT_DATA, "Arrays should have same length");
}
}
index = std::make_shared<AnnoyIndex>(size);
for (size_t current_row = 0; current_row < num_rows; ++current_row)
{
index->add_item(index->get_n_items(), &array[offsets[current_row]]);
}
}
else
{
/// Other possible type of column is Tuple
const auto & column_tuple = typeid_cast<const ColumnTuple*>(column_cut.get());
if (!column_tuple)
throw Exception(ErrorCodes::INCORRECT_QUERY, "Wrong type was given to index.");
const auto & columns = column_tuple->getColumns();
std::vector<std::vector<Float32>> data{column_tuple->size(), std::vector<Float32>()};
for (const auto& column : columns)
{
const auto& pod_array = typeid_cast<const ColumnFloat32*>(column.get())->getData();
for (size_t i = 0; i < pod_array.size(); ++i)
{
data[i].push_back(pod_array[i]);
}
}
assert(!data.empty());
if (!index)
{
index = std::make_shared<AnnoyIndex>(data[0].size());
}
for (const auto& item : data)
{
index->add_item(index->get_n_items(), item.data());
}
}
*pos += rows_read;
}
MergeTreeIndexConditionAnnoy::MergeTreeIndexConditionAnnoy(
const IndexDescription & /*index*/,
const SelectQueryInfo & query,
ContextPtr context)
: condition(query, context)
{}
bool MergeTreeIndexConditionAnnoy::mayBeTrueOnGranule(MergeTreeIndexGranulePtr /* idx_granule */) const
{
throw Exception("mayBeTrueOnGranule is not supported for ANN skip indexes", ErrorCodes::LOGICAL_ERROR);
}
bool MergeTreeIndexConditionAnnoy::alwaysUnknownOrTrue() const
{
return condition.alwaysUnknownOrTrue("L2Distance");
}
std::vector<size_t> MergeTreeIndexConditionAnnoy::getUsefulRanges(MergeTreeIndexGranulePtr idx_granule) const
{
UInt64 limit = condition.getLimit();
UInt64 index_granularity = condition.getIndexGranularity();
std::optional<float> comp_dist = condition.getQueryType() == ANN::ANNQueryInformation::Type::Where ?
std::optional<float>(condition.getComparisonDistanceForWhereQuery()) : std::nullopt;
if (comp_dist && comp_dist.value() < 0)
throw Exception(ErrorCodes::LOGICAL_ERROR, "Attempt to optimize query with where without distance");
std::vector<float> target_vec = condition.getTargetVector();
auto granule = std::dynamic_pointer_cast<MergeTreeIndexGranuleAnnoy>(idx_granule);
if (granule == nullptr)
{
throw Exception("Granule has the wrong type", ErrorCodes::LOGICAL_ERROR);
}
auto annoy = granule->index;
if (condition.getNumOfDimensions() != annoy->getNumOfDimensions())
{
throw Exception("The dimension of the space in the request (" + toString(condition.getNumOfDimensions()) + ") "
+ "does not match with the dimension in the index (" + toString(annoy->getNumOfDimensions()) + ")", ErrorCodes::INCORRECT_QUERY);
}
/// neighbors contain indexes of dots which were closest to target vector
std::vector<UInt64> neighbors;
std::vector<Float32> distances;
neighbors.reserve(limit);
distances.reserve(limit);
int k_search = -1;
String params_str = condition.getParamsStr();
if (!params_str.empty())
{
try
{
/// k_search=... (algorithm will inspect up to search_k nodes which defaults to n_trees * n if not provided)
k_search = std::stoi(params_str.data() + 9);
}
catch (...)
{
throw Exception("Setting of the annoy index should be int", ErrorCodes::INCORRECT_QUERY);
}
}
annoy->get_nns_by_vector(target_vec.data(), limit, k_search, &neighbors, &distances);
std::unordered_set<size_t> granule_numbers;
for (size_t i = 0; i < neighbors.size(); ++i)
{
if (comp_dist && distances[i] > comp_dist)
{
continue;
}
granule_numbers.insert(neighbors[i] / index_granularity);
}
std::vector<size_t> result_vector;
result_vector.reserve(granule_numbers.size());
for (auto granule_number : granule_numbers)
{
result_vector.push_back(granule_number);
}
return result_vector;
}
MergeTreeIndexGranulePtr MergeTreeIndexAnnoy::createIndexGranule() const
{
return std::make_shared<MergeTreeIndexGranuleAnnoy>(index.name, index.sample_block);
}
MergeTreeIndexAggregatorPtr MergeTreeIndexAnnoy::createIndexAggregator() const
{
return std::make_shared<MergeTreeIndexAggregatorAnnoy>(index.name, index.sample_block, number_of_trees);
}
MergeTreeIndexConditionPtr MergeTreeIndexAnnoy::createIndexCondition(
const SelectQueryInfo & query, ContextPtr context) const
{
return std::make_shared<MergeTreeIndexConditionAnnoy>(index, query, context);
};
MergeTreeIndexPtr annoyIndexCreator(const IndexDescription & index)
{
uint64_t param = index.arguments[0].get<uint64_t>();
return std::make_shared<MergeTreeIndexAnnoy>(index, param);
}
void annoyIndexValidator(const IndexDescription & index, bool /* attach */)
{
if (index.arguments.size() != 1)
{
throw Exception("Annoy index must have exactly one argument.", ErrorCodes::INCORRECT_QUERY);
}
if (index.arguments[0].getType() != Field::Types::UInt64)
{
throw Exception("Annoy index argument must be UInt64.", ErrorCodes::INCORRECT_QUERY);
}
}
}
#endif // ENABLE_ANNOY

123
src/Storages/MergeTree/MergeTreeIndexAnnoy.h
View File

@ -1,123 +0,0 @@
#pragma once
#ifdef ENABLE_ANNOY
#include <Storages/MergeTree/CommonANNIndexes.h>
#include <annoylib.h>
#include <kissrandom.h>
namespace DB
{
namespace ANN = ApproximateNearestNeighbour;
// auxiliary namespace for working with spotify-annoy library
// mainly for serialization and deserialization of the index
namespace ApproximateNearestNeighbour
{
using AnnoyIndexThreadedBuildPolicy = ::Annoy::AnnoyIndexMultiThreadedBuildPolicy;
// TODO: Support different metrics. List of available metrics can be taken from here:
// https://github.com/spotify/annoy/blob/master/src/annoymodule.cc#L151-L171
template <typename Distance = ::Annoy::Euclidean>
class AnnoyIndex : public ::Annoy::AnnoyIndex<UInt64, Float32, Distance, ::Annoy::Kiss64Random, AnnoyIndexThreadedBuildPolicy>
{
using Base = ::Annoy::AnnoyIndex<UInt64, Float32, Distance, ::Annoy::Kiss64Random, AnnoyIndexThreadedBuildPolicy>;
public:
explicit AnnoyIndex(const uint64_t dim) : Base::AnnoyIndex(dim) {}
void serialize(WriteBuffer& ostr) const;
void deserialize(ReadBuffer& istr);
uint64_t getNumOfDimensions() const;
};
}
struct MergeTreeIndexGranuleAnnoy final : public IMergeTreeIndexGranule
{
using AnnoyIndex = ANN::AnnoyIndex<>;
using AnnoyIndexPtr = std::shared_ptr<AnnoyIndex>;
MergeTreeIndexGranuleAnnoy(const String & index_name_, const Block & index_sample_block_);
MergeTreeIndexGranuleAnnoy(
const String & index_name_,
const Block & index_sample_block_,
AnnoyIndexPtr index_base_);
~MergeTreeIndexGranuleAnnoy() override = default;
void serializeBinary(WriteBuffer & ostr) const override;
void deserializeBinary(ReadBuffer & istr, MergeTreeIndexVersion version) override;
bool empty() const override { return !index.get(); }
String index_name;
Block index_sample_block;
AnnoyIndexPtr index;
};
struct MergeTreeIndexAggregatorAnnoy final : IMergeTreeIndexAggregator
{
using AnnoyIndex = ANN::AnnoyIndex<>;
using AnnoyIndexPtr = std::shared_ptr<AnnoyIndex>;
MergeTreeIndexAggregatorAnnoy(const String & index_name_, const Block & index_sample_block, uint64_t number_of_trees);
~MergeTreeIndexAggregatorAnnoy() override = default;
bool empty() const override { return !index || index->get_n_items() == 0; }
MergeTreeIndexGranulePtr getGranuleAndReset() override;
void update(const Block & block, size_t * pos, size_t limit) override;
String index_name;
Block index_sample_block;
const uint64_t number_of_trees;
AnnoyIndexPtr index;
};
class MergeTreeIndexConditionAnnoy final : public ANN::IMergeTreeIndexConditionAnn
{
public:
MergeTreeIndexConditionAnnoy(
const IndexDescription & index,
const SelectQueryInfo & query,
ContextPtr context);
bool alwaysUnknownOrTrue() const override;
bool mayBeTrueOnGranule(MergeTreeIndexGranulePtr idx_granule) const override;
std::vector<size_t> getUsefulRanges(MergeTreeIndexGranulePtr idx_granule) const override;
~MergeTreeIndexConditionAnnoy() override = default;
private:
ANN::ANNCondition condition;
};
class MergeTreeIndexAnnoy : public IMergeTreeIndex
{
public:
MergeTreeIndexAnnoy(const IndexDescription & index_, uint64_t number_of_trees_)
: IMergeTreeIndex(index_)
, number_of_trees(number_of_trees_)
{}
~MergeTreeIndexAnnoy() override = default;
MergeTreeIndexGranulePtr createIndexGranule() const override;
MergeTreeIndexAggregatorPtr createIndexAggregator() const override;
MergeTreeIndexConditionPtr createIndexCondition(
const SelectQueryInfo & query, ContextPtr context) const override;
bool mayBenefitFromIndexForIn(const ASTPtr & /*node*/) const override { return false; }
private:
const uint64_t number_of_trees;
};
}
#endif // ENABLE_ANNOY

5
src/Storages/MergeTree/MergeTreeIndices.cpp
View File

@ -101,11 +101,6 @@ MergeTreeIndexFactory::MergeTreeIndexFactory()
registerCreator("hypothesis", hypothesisIndexCreator);
registerValidator("hypothesis", hypothesisIndexValidator);
#ifdef ENABLE_ANNOY
registerCreator("annoy", annoyIndexCreator);
registerValidator("annoy", annoyIndexValidator);
#endif
}
MergeTreeIndexFactory & MergeTreeIndexFactory::instance()

5
src/Storages/MergeTree/MergeTreeIndices.h
View File

@ -224,9 +224,4 @@ void bloomFilterIndexValidatorNew(const IndexDescription & index, bool attach);
MergeTreeIndexPtr hypothesisIndexCreator(const IndexDescription & index);
void hypothesisIndexValidator(const IndexDescription & index, bool attach);
#ifdef ENABLE_ANNOY
MergeTreeIndexPtr annoyIndexCreator(const IndexDescription & index);
void annoyIndexValidator(const IndexDescription & index, bool attach);
#endif
}

10
src/Storages/MySQL/MySQLHelpers.cpp
View File

@ -4,6 +4,7 @@
#include <mysqlxx/PoolWithFailover.h>
#include <Storages/ExternalDataSourceConfiguration.h>
#include <Storages/MySQL/MySQLSettings.h>
#include <Databases/MySQL/ConnectionMySQLSettings.h>
namespace DB
{
@ -13,8 +14,8 @@ namespace ErrorCodes
extern const int BAD_ARGUMENTS;
}
mysqlxx::PoolWithFailover
createMySQLPoolWithFailover(const StorageMySQLConfiguration & configuration, const MySQLSettings & mysql_settings)
template <typename T> mysqlxx::PoolWithFailover
createMySQLPoolWithFailover(const StorageMySQLConfiguration & configuration, const T & mysql_settings)
{
if (!mysql_settings.connection_pool_size)
throw Exception(ErrorCodes::BAD_ARGUMENTS, "Connection pool cannot have zero size");
@ -29,6 +30,11 @@ createMySQLPoolWithFailover(const StorageMySQLConfiguration & configuration, con
mysql_settings.read_write_timeout);
}
template
mysqlxx::PoolWithFailover createMySQLPoolWithFailover(const StorageMySQLConfiguration & configuration, const MySQLSettings & mysql_settings);
template
mysqlxx::PoolWithFailover createMySQLPoolWithFailover(const StorageMySQLConfiguration & configuration, const ConnectionMySQLSettings & mysql_settings);
}
#endif

5
src/Storages/MySQL/MySQLHelpers.h
View File

@ -9,10 +9,9 @@ namespace mysqlxx { class PoolWithFailover; }
namespace DB
{
struct StorageMySQLConfiguration;
struct MySQLSettings;
mysqlxx::PoolWithFailover
createMySQLPoolWithFailover(const StorageMySQLConfiguration & configuration, const MySQLSettings & mysql_settings);
template <typename T> mysqlxx::PoolWithFailover
createMySQLPoolWithFailover(const StorageMySQLConfiguration & configuration, const T & mysql_settings);
}

8
src/Storages/MySQL/MySQLSettings.cpp
View File

@ -15,13 +15,18 @@ namespace ErrorCodes
IMPLEMENT_SETTINGS_TRAITS(MySQLSettingsTraits, LIST_OF_MYSQL_SETTINGS)
void MySQLSettings::loadFromQuery(const ASTSetQuery & settings_def)
{
applyChanges(settings_def.changes);
}
void MySQLSettings::loadFromQuery(ASTStorage & storage_def)
{
if (storage_def.settings)
{
try
{
applyChanges(storage_def.settings->changes);
loadFromQuery(*storage_def.settings);
}
catch (Exception & e)
{
@ -39,4 +44,3 @@ void MySQLSettings::loadFromQuery(ASTStorage & storage_def)
}
}

2
src/Storages/MySQL/MySQLSettings.h
View File

@ -13,6 +13,7 @@ namespace Poco::Util
namespace DB
{
class ASTStorage;
class ASTSetQuery;
#define LIST_OF_MYSQL_SETTINGS(M) \
M(UInt64, connection_pool_size, 16, "Size of connection pool (if all connections are in use, the query will wait until some connection will be freed).", 0) \
@ -32,6 +33,7 @@ using MySQLBaseSettings = BaseSettings<MySQLSettingsTraits>;
struct MySQLSettings : public MySQLBaseSettings
{
void loadFromQuery(ASTStorage & storage_def);
void loadFromQuery(const ASTSetQuery & settings_def);
};

17
src/TableFunctions/TableFunctionMySQL.cpp
View File

@ -37,11 +37,26 @@ void TableFunctionMySQL::parseArguments(const ASTPtr & ast_function, ContextPtr
if (!args_func.arguments)
throw Exception("Table function 'mysql' must have arguments.", ErrorCodes::LOGICAL_ERROR);
auto & args = args_func.arguments->children;
MySQLSettings mysql_settings;
configuration = StorageMySQL::getConfiguration(args_func.arguments->children, context, mysql_settings);
const auto & settings = context->getSettingsRef();
mysql_settings.connect_timeout = settings.external_storage_connect_timeout_sec;
mysql_settings.read_write_timeout = settings.external_storage_rw_timeout_sec;
for (auto it = args.begin(); it != args.end(); ++it)
{
const ASTSetQuery * settings_ast = (*it)->as<ASTSetQuery>();
if (settings_ast)
{
mysql_settings.loadFromQuery(*settings_ast);
args.erase(it);
break;
}
}
configuration = StorageMySQL::getConfiguration(args, context, mysql_settings);
pool.emplace(createMySQLPoolWithFailover(*configuration, mysql_settings));
}

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

11
tests/integration/test_host_regexp_hosts_file_resolution/configs/host_regexp.xml Normal file
View File

@ -0,0 +1,11 @@
<yandex>
<users>
<test_dns>
<password/>
<networks>
<host_regexp>test1\.example\.com$</host_regexp>
</networks>
<profile>default</profile>
</test_dns>
</users>
</yandex>

5
tests/integration/test_host_regexp_hosts_file_resolution/configs/listen_host.xml Normal file
View File

@ -0,0 +1,5 @@
<yandex>
<listen_host>::</listen_host>
<listen_host>0.0.0.0</listen_host>
<listen_try>1</listen_try>
</yandex>

46
tests/integration/test_host_regexp_hosts_file_resolution/test.py Normal file
View File

@ -0,0 +1,46 @@
import pytest
from helpers.cluster import ClickHouseCluster, get_docker_compose_path, run_and_check
import os
DOCKER_COMPOSE_PATH = get_docker_compose_path()
SCRIPT_DIR = os.path.dirname(os.path.realpath(__file__))
cluster = ClickHouseCluster(__file__)
ch_server = cluster.add_instance(
"clickhouse-server",
main_configs=["configs/listen_host.xml"],
user_configs=["configs/host_regexp.xml"],
)
client = cluster.add_instance(
"clickhouse-client",
)
def build_endpoint_v4(ip):
return f"'http://{ip}:8123/?query=SELECT+1&user=test_dns'"
@pytest.fixture(scope="module")
def started_cluster():
global cluster
try:
cluster.start()
yield cluster
finally:
cluster.shutdown()
def test_host_regexp_multiple_ptr_hosts_file_v4(started_cluster):
server_ip = cluster.get_instance_ip("clickhouse-server")
client_ip = cluster.get_instance_ip("clickhouse-client")
ch_server.exec_in_container(
(["bash", "-c", f"echo '{client_ip} test1.example.com' > /etc/hosts"])
)
endpoint = build_endpoint_v4(server_ip)
assert "1\n" == client.exec_in_container((["bash", "-c", f"curl {endpoint}"]))

2
tests/integration/test_mysql_protocol/test.py
View File

@ -250,7 +250,7 @@ def test_mysql_client_exception(started_cluster):
expected_msg = "\n".join(
[
"mysql: [Warning] Using a password on the command line interface can be insecure.",
"ERROR 1000 (00000) at line 1: Poco::Exception. Code: 1000, e.code() = 0, Exception: Connections to all replicas failed: default@127.0.0.1:10086 as user default",
"ERROR 1000 (00000) at line 1: Poco::Exception. Code: 1000, e.code() = 0, Exception: Connections to mysql failed: default@127.0.0.1:10086 as user default",
]
)
assert stderr[: len(expected_msg)].decode() == expected_msg

8
tests/integration/test_replicated_merge_tree_s3_zero_copy/configs/config.d/storage_conf.xml
View File

@ -7,12 +7,18 @@
<access_key_id>minio</access_key_id>
<secret_access_key>minio123</secret_access_key>
</s3>
<cache_s3>
<type>cache</type>
<disk>s3</disk>
<max_size>100000000</max_size>
<path>./cache_s3/</path>
</cache_s3>
</disks>
<policies>
<s3>
<volumes>
<main>
<disk>s3</disk>
<disk>cache_s3</disk>
</main>
</volumes>
</s3>

11
tests/integration/test_storage_mysql/configs/named_collections.xml
View File

@ -30,5 +30,16 @@
<table>test_table</table>
<connection_pool_size>0</connection_pool_size>
</mysql4>
<mysql_with_settings>
<user>root</user>
<password>clickhouse</password>
<host>mysql57</host>
<port>3306</port>
<database>clickhouse</database>
<table>test_settings</table>
<connection_pool_size>1</connection_pool_size>
<read_write_timeout>20123001</read_write_timeout>
<connect_timeout>20123002</connect_timeout>
</mysql_with_settings>
</named_collections>
</clickhouse>

87
tests/integration/test_storage_mysql/test.py
View File

@ -732,6 +732,93 @@ def test_mysql_null(started_cluster):
conn.close()
def test_settings(started_cluster):
table_name = "test_settings"
node1.query(f"DROP TABLE IF EXISTS {table_name}")
wait_timeout = 123
rw_timeout = 10123001
connect_timeout = 10123002
connection_pool_size = 1
conn = get_mysql_conn(started_cluster, cluster.mysql_ip)
drop_mysql_table(conn, table_name)
create_mysql_table(conn, table_name)
node1.query(
f"""
CREATE TABLE {table_name}
(
id UInt32,
name String,
age UInt32,
money UInt32
)
ENGINE = MySQL('mysql57:3306', 'clickhouse', '{table_name}', 'root', 'clickhouse')
SETTINGS connection_wait_timeout={wait_timeout}, connect_timeout={connect_timeout}, read_write_timeout={rw_timeout}, connection_pool_size={connection_pool_size}
"""
)
node1.query(f"SELECT * FROM {table_name}")
assert node1.contains_in_log(
f"with settings: connect_timeout={connect_timeout}, read_write_timeout={rw_timeout}"
)
rw_timeout = 20123001
connect_timeout = 20123002
node1.query(f"SELECT * FROM mysql(mysql_with_settings)")
assert node1.contains_in_log(
f"with settings: connect_timeout={connect_timeout}, read_write_timeout={rw_timeout}"
)
rw_timeout = 30123001
connect_timeout = 30123002
node1.query(
f"""
SELECT *
FROM mysql('mysql57:3306', 'clickhouse', '{table_name}', 'root', 'clickhouse',
SETTINGS
connection_wait_timeout={wait_timeout},
connect_timeout={connect_timeout},
read_write_timeout={rw_timeout},
connection_pool_size={connection_pool_size})
"""
)
assert node1.contains_in_log(
f"with settings: connect_timeout={connect_timeout}, read_write_timeout={rw_timeout}"
)
rw_timeout = 40123001
connect_timeout = 40123002
node1.query(
f"""
CREATE DATABASE m
ENGINE = MySQL(mysql_with_settings, connection_wait_timeout={wait_timeout}, connect_timeout={connect_timeout}, read_write_timeout={rw_timeout}, connection_pool_size={connection_pool_size})
"""
)
assert node1.contains_in_log(
f"with settings: connect_timeout={connect_timeout}, read_write_timeout={rw_timeout}"
)
rw_timeout = 50123001
connect_timeout = 50123002
node1.query(
f"""
CREATE DATABASE mm ENGINE = MySQL('mysql57:3306', 'clickhouse', 'root', 'clickhouse')
SETTINGS
connection_wait_timeout={wait_timeout},
connect_timeout={connect_timeout},
read_write_timeout={rw_timeout},
connection_pool_size={connection_pool_size}
"""
)
assert node1.contains_in_log(
f"with settings: connect_timeout={connect_timeout}, read_write_timeout={rw_timeout}"
)
drop_mysql_table(conn, table_name)
conn.close()
if __name__ == "__main__":
with contextmanager(started_cluster)() as cluster:
for name, instance in list(cluster.instances.items()):

16
tests/queries/0_stateless/02354_annoy.reference
View File

@ -1,16 +0,0 @@
1 [0,0,10]
2 [0,0,10.5]
3 [0,0,9.5]
4 [0,0,9.7]
5 [0,0,10.2]
1 [0,0,10]
5 [0,0,10.2]
4 [0,0,9.7]
1 [0,0,10]
2 [0,0,10.5]
3 [0,0,9.5]
4 [0,0,9.7]
5 [0,0,10.2]
1 [0,0,10]
5 [0,0,10.2]
4 [0,0,9.7]

44
tests/queries/0_stateless/02354_annoy.sql
View File

@ -1,44 +0,0 @@
-- Tags: no-fasttest, no-ubsan
DROP TABLE IF EXISTS 02354_annoy;
CREATE TABLE 02354_annoy
(
id Int32,
embedding Array(Float32),
INDEX annoy_index embedding TYPE annoy(100) GRANULARITY 1
)
ENGINE = MergeTree
ORDER BY id
SETTINGS index_granularity=5;
INSERT INTO 02354_annoy VALUES (1, [0.0, 0.0, 10.0]), (2, [0.0, 0.0, 10.5]), (3, [0.0, 0.0, 9.5]), (4, [0.0, 0.0, 9.7]), (5, [0.0, 0.0, 10.2]), (6, [10.0, 0.0, 0.0]), (7, [9.5, 0.0, 0.0]), (8, [9.7, 0.0, 0.0]), (9, [10.2, 0.0, 0.0]), (10, [10.5, 0.0, 0.0]), (11, [0.0, 10.0, 0.0]), (12, [0.0, 9.5, 0.0]), (13, [0.0, 9.7, 0.0]), (14, [0.0, 10.2, 0.0]), (15, [0.0, 10.5, 0.0]);
SELECT *
FROM 02354_annoy
WHERE L2Distance(embedding, [0.0, 0.0, 10.0]) < 1.0
LIMIT 5;
SELECT *
FROM 02354_annoy
ORDER BY L2Distance(embedding, [0.0, 0.0, 10.0])
LIMIT 3;
SET param_02354_target_vector='[0.0, 0.0, 10.0]';
SELECT *
FROM 02354_annoy
WHERE L2Distance(embedding, {02354_target_vector: Array(Float32)}) < 1.0
LIMIT 5;
SELECT *
FROM 02354_annoy
ORDER BY L2Distance(embedding, {02354_target_vector: Array(Float32)})
LIMIT 3;
SELECT *
FROM 02354_annoy
ORDER BY L2Distance(embedding, [0.0, 0.0])
LIMIT 3; -- { serverError 80 }
DROP TABLE IF EXISTS 02354_annoy;

1
tests/queries/0_stateless/02354_with_statement_non_exist_column.reference Normal file
View File

@ -0,0 +1 @@
1

9
tests/queries/0_stateless/02354_with_statement_non_exist_column.sql Normal file
View File

@ -0,0 +1,9 @@
WITH x AS y SELECT 1;
DROP TEMPORARY TABLE IF EXISTS t1;
DROP TEMPORARY TABLE IF EXISTS t2;
CREATE TEMPORARY TABLE t1 (a Int64);
CREATE TEMPORARY TABLE t2 (a Int64, b Int64);
WITH b AS bb SELECT bb FROM t2 WHERE a IN (SELECT a FROM t1);