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

Merge with master

Browse Source
This commit is contained in:
alesapin 2019-08-21 11:57:19 +03:00
commit 70ca7f4156
283 changed files with 6529 additions and 890 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
.github/label-pr.yml vendored Normal file
View File

@ -0,0 +1,2 @@
- regExp: ".*\\.md$"
labels: ["documentation", "pr-documentation"]

9
.github/main.workflow vendored Normal file
View File

@ -0,0 +1,9 @@
workflow "Main workflow" {
resolves = ["Label PR"]
on = "pull_request"
}
action "Label PR" {
uses = "decathlon/pull-request-labeler-action@v1.0.0"
secrets = ["GITHUB_TOKEN"]
}

55
CHANGELOG.md
View File

@ -1,3 +1,56 @@
## ClickHouse release 19.13.2.19, 2019-08-14
### New Feature
* Sampling profiler on query level. [Example](https://gist.github.com/alexey-milovidov/92758583dd41c24c360fdb8d6a4da194). [#4247](https://github.com/yandex/ClickHouse/issues/4247) ([laplab](https://github.com/laplab)) [#6124](https://github.com/yandex/ClickHouse/pull/6124) ([alexey-milovidov](https://github.com/alexey-milovidov)) [#6250](https://github.com/yandex/ClickHouse/pull/6250) [#6283](https://github.com/yandex/ClickHouse/pull/6283) [#6386](https://github.com/yandex/ClickHouse/pull/6386)
* Allow to specify a list of columns with `COLUMNS('regexp')` expression that works like a more sophisticated variant of `*` asterisk. [#5951](https://github.com/yandex/ClickHouse/pull/5951) ([mfridental](https://github.com/mfridental)), ([alexey-milovidov](https://github.com/alexey-milovidov))
* `CREATE TABLE AS table_function()` is now possible [#6057](https://github.com/yandex/ClickHouse/pull/6057) ([dimarub2000](https://github.com/dimarub2000))
* Adam optimizer for stochastic gradient descent is used by default in `stochasticLinearRegression()` and `stochasticLogisticRegression()` aggregate functions, because it shows good quality without almost any tuning. [#6000](https://github.com/yandex/ClickHouse/pull/6000) ([Quid37](https://github.com/Quid37))
* Added functions for working with the сustom week number [#5212](https://github.com/yandex/ClickHouse/pull/5212) ([Andy Yang](https://github.com/andyyzh))
* `RENAME` queries now work with all storages. [#5953](https://github.com/yandex/ClickHouse/pull/5953) ([Ivan](https://github.com/abyss7))
* Now client receive logs from server with any desired level by setting `send_logs_level` regardless to the log level specified in server settings. [#5964](https://github.com/yandex/ClickHouse/pull/5964) ([Nikita Mikhaylov](https://github.com/nikitamikhaylov))
### Backward Incompatible Change
* The setting `input_format_defaults_for_omitted_fields` is enabled by default. Inserts in Distibuted tables need this setting to be the same on cluster (you need to set it before rolling update). It enables calculation of complex default expressions for omitted fields in `JSONEachRow` and `CSV*` formats. It should be the expected behaviour but may lead to negligible performance difference. [#6043](https://github.com/yandex/ClickHouse/pull/6043) ([Artem Zuikov](https://github.com/4ertus2)), [#5625](https://github.com/yandex/ClickHouse/pull/5625) ([akuzm](https://github.com/akuzm))
### Experimental features
* New query processing pipeline. Use `experimental_use_processors=1` option to enable it. Use for your own trouble. [#4914](https://github.com/yandex/ClickHouse/pull/4914) ([Nikolai Kochetov](https://github.com/KochetovNicolai))
### Bug Fix
* Kafka integration has been fixed in this version.
* Fixed `DoubleDelta` encoding of `Int64` for large `DoubleDelta` values, improved `DoubleDelta` encoding for random data for `Int32`. [#5998](https://github.com/yandex/ClickHouse/pull/5998) ([Vasily Nemkov](https://github.com/Enmk))
* Fixed overestimation of `max_rows_to_read` if the setting `merge_tree_uniform_read_distribution` is set to 0. [#6019](https://github.com/yandex/ClickHouse/pull/6019) ([alexey-milovidov](https://github.com/alexey-milovidov))
### Improvement
* Throws an exception if `config.d` file doesn't have the corresponding root element as the config file [#6123](https://github.com/yandex/ClickHouse/pull/6123) ([dimarub2000](https://github.com/dimarub2000))
### Performance Improvement
* Optimize `count()`. Now it uses the smallest column (if possible). [#6028](https://github.com/yandex/ClickHouse/pull/6028) ([Amos Bird](https://github.com/amosbird))
### Build/Testing/Packaging Improvement
* Report memory usage in performance tests. [#5899](https://github.com/yandex/ClickHouse/pull/5899) ([akuzm](https://github.com/akuzm))
* Fix build with external `libcxx` [#6010](https://github.com/yandex/ClickHouse/pull/6010) ([Ivan](https://github.com/abyss7))
* Fix shared build with `rdkafka` library [#6101](https://github.com/yandex/ClickHouse/pull/6101) ([Ivan](https://github.com/abyss7))
## ClickHouse release 19.11.7.40, 2019-08-14
### Bug fix
* Kafka integration has been fixed in this version.
* Fix segfault when using `arrayReduce` for constant arguments. [#6326](https://github.com/yandex/ClickHouse/pull/6326) ([alexey-milovidov](https://github.com/alexey-milovidov))
* Fixed `toFloat()` monotonicity. [#6374](https://github.com/yandex/ClickHouse/pull/6374) ([dimarub2000](https://github.com/dimarub2000))
* Fix segfault with enabled `optimize_skip_unused_shards` and missing sharding key. [#6384](https://github.com/yandex/ClickHouse/pull/6384) ([CurtizJ](https://github.com/CurtizJ))
* Fixed logic of `arrayEnumerateUniqRanked` function. [#6423](https://github.com/yandex/ClickHouse/pull/6423) ([alexey-milovidov](https://github.com/alexey-milovidov))
* Removed extra verbose logging from MySQL handler. [#6389](https://github.com/yandex/ClickHouse/pull/6389) ([alexey-milovidov](https://github.com/alexey-milovidov))
* Fix wrong behavior and possible segfaults in `topK` and `topKWeighted` aggregated functions. [#6404](https://github.com/yandex/ClickHouse/pull/6404) ([CurtizJ](https://github.com/CurtizJ))
* Do not expose virtual columns in `system.columns` table. This is required for backward compatibility. [#6406](https://github.com/yandex/ClickHouse/pull/6406) ([alexey-milovidov](https://github.com/alexey-milovidov))
* Fix bug with memory allocation for string fields in complex key cache dictionary. [#6447](https://github.com/yandex/ClickHouse/pull/6447) ([alesapin](https://github.com/alesapin))
* Fix bug with enabling adaptive granularity when creating new replica for `Replicated*MergeTree` table. [#6452](https://github.com/yandex/ClickHouse/pull/6452) ([alesapin](https://github.com/alesapin))
* Fix infinite loop when reading Kafka messages. [#6354](https://github.com/yandex/ClickHouse/pull/6354) ([abyss7](https://github.com/abyss7))
* Fixed the possibility of a fabricated query to cause server crash due to stack overflow in SQL parser and possibility of stack overflow in `Merge` and `Distributed` tables [#6433](https://github.com/yandex/ClickHouse/pull/6433) ([alexey-milovidov](https://github.com/alexey-milovidov))
* Fixed Gorilla encoding error on small sequences. [#6444](https://github.com/yandex/ClickHouse/pull/6444) ([Enmk](https://github.com/Enmk))
### Improvement
* Allow user to override `poll_interval` and `idle_connection_timeout` settings on connection. [#6230](https://github.com/yandex/ClickHouse/pull/6230) ([alexey-milovidov](https://github.com/alexey-milovidov))
## ClickHouse release 19.11.5.28, 2019-08-05
### Bug fix
@ -299,7 +352,7 @@ It allows to set commit mode: after every batch of messages is handled, or after
* Renamed functions `leastSqr` to `simpleLinearRegression`, `LinearRegression` to `linearRegression`, `LogisticRegression` to `logisticRegression`. [#5391](https://github.com/yandex/ClickHouse/pull/5391) ([Nikolai Kochetov](https://github.com/KochetovNicolai))
### Performance Improvements
* Paralellize processing of parts in alter modify query. [#4639](https://github.com/yandex/ClickHouse/pull/4639) ([Ivan Kush](https://github.com/IvanKush))
* Paralellize processing of parts of non-replicated MergeTree tables in ALTER MODIFY query. [#4639](https://github.com/yandex/ClickHouse/pull/4639) ([Ivan Kush](https://github.com/IvanKush))
* Optimizations in regular expressions extraction. [#5193](https://github.com/yandex/ClickHouse/pull/5193) [#5191](https://github.com/yandex/ClickHouse/pull/5191) ([Danila Kutenin](https://github.com/danlark1))
* Do not add right join key column to join result if it's used only in join on section. [#5260](https://github.com/yandex/ClickHouse/pull/5260) ([Artem Zuikov](https://github.com/4ertus2))
* Freeze the Kafka buffer after first empty response. It avoids multiple invokations of `ReadBuffer::next()` for empty result in some row-parsing streams. [#5283](https://github.com/yandex/ClickHouse/pull/5283) ([Ivan](https://github.com/abyss7))

2
cmake/find_parquet.cmake
View File

@ -2,7 +2,7 @@ option (ENABLE_PARQUET "Enable parquet" ON)
if (ENABLE_PARQUET)
if (NOT OS_FREEBSD) # Freebsd: ../contrib/arrow/cpp/src/arrow/util/bit-util.h:27:10: fatal error: endian.h: No such file or directory
if (NOT OS_FREEBSD AND NOT APPLE) # Freebsd: ../contrib/arrow/cpp/src/arrow/util/bit-util.h:27:10: fatal error: endian.h: No such file or directory
option(USE_INTERNAL_PARQUET_LIBRARY "Set to FALSE to use system parquet library instead of bundled" ${NOT_UNBUNDLED})
endif()

16
dbms/programs/copier/ClusterCopier.cpp
View File

@ -1932,15 +1932,13 @@ protected:
TaskTable & task_table = task_shard.task_table;
String query;
{
WriteBufferFromOwnString wb;
wb << "SELECT 1"
<< " FROM "<< getQuotedTable(task_shard.table_read_shard)
<< " WHERE " << queryToString(task_table.engine_push_partition_key_ast) << " = " << partition_quoted_name
<< " LIMIT 1";
query = wb.str();
}
std::string query = "SELECT 1 FROM " + getQuotedTable(task_shard.table_read_shard)
+ " WHERE (" + queryToString(task_table.engine_push_partition_key_ast) + " = (" + partition_quoted_name + " AS partition_key))";
if (!task_table.where_condition_str.empty())
query += " AND (" + task_table.where_condition_str + ")";
query += " LIMIT 1";
LOG_DEBUG(log, "Checking shard " << task_shard.getDescription() << " for partition "
<< partition_quoted_name << " existence, executing query: " << query);

1
dbms/programs/odbc-bridge/CMakeLists.txt
View File

@ -5,7 +5,6 @@ set(CLICKHOUSE_ODBC_BRIDGE_SOURCES
${CMAKE_CURRENT_SOURCE_DIR}/IdentifierQuoteHandler.cpp
${CMAKE_CURRENT_SOURCE_DIR}/MainHandler.cpp
${CMAKE_CURRENT_SOURCE_DIR}/ODBCBlockInputStream.cpp
${CMAKE_CURRENT_SOURCE_DIR}/odbc-bridge.cpp
${CMAKE_CURRENT_SOURCE_DIR}/ODBCBridge.cpp
${CMAKE_CURRENT_SOURCE_DIR}/PingHandler.cpp
${CMAKE_CURRENT_SOURCE_DIR}/validateODBCConnectionString.cpp

12
dbms/src/AggregateFunctions/AggregateFunctionQuantile.cpp
View File

@ -24,6 +24,12 @@ template <typename Value, bool FloatReturn> using FuncQuantilesDeterministic = A
template <typename Value, bool _> using FuncQuantileExact = AggregateFunctionQuantile<Value, QuantileExact<Value>, NameQuantileExact, false, void, false>;
template <typename Value, bool _> using FuncQuantilesExact = AggregateFunctionQuantile<Value, QuantileExact<Value>, NameQuantilesExact, false, void, true>;
template <typename Value, bool _> using FuncQuantileExactExclusive = AggregateFunctionQuantile<Value, QuantileExactExclusive<Value>, NameQuantileExactExclusive, false, Float64, false>;
template <typename Value, bool _> using FuncQuantilesExactExclusive = AggregateFunctionQuantile<Value, QuantileExactExclusive<Value>, NameQuantilesExactExclusive, false, Float64, true>;
template <typename Value, bool _> using FuncQuantileExactInclusive = AggregateFunctionQuantile<Value, QuantileExactInclusive<Value>, NameQuantileExactInclusive, false, Float64, false>;
template <typename Value, bool _> using FuncQuantilesExactInclusive = AggregateFunctionQuantile<Value, QuantileExactInclusive<Value>, NameQuantilesExactInclusive, false, Float64, true>;
template <typename Value, bool _> using FuncQuantileExactWeighted = AggregateFunctionQuantile<Value, QuantileExactWeighted<Value>, NameQuantileExactWeighted, true, void, false>;
template <typename Value, bool _> using FuncQuantilesExactWeighted = AggregateFunctionQuantile<Value, QuantileExactWeighted<Value>, NameQuantilesExactWeighted, true, void, true>;
@ -92,6 +98,12 @@ void registerAggregateFunctionsQuantile(AggregateFunctionFactory & factory)
factory.registerFunction(NameQuantileExact::name, createAggregateFunctionQuantile<FuncQuantileExact>);
factory.registerFunction(NameQuantilesExact::name, createAggregateFunctionQuantile<FuncQuantilesExact>);
factory.registerFunction(NameQuantileExactExclusive::name, createAggregateFunctionQuantile<FuncQuantileExactExclusive>);
factory.registerFunction(NameQuantilesExactExclusive::name, createAggregateFunctionQuantile<FuncQuantilesExactExclusive>);
factory.registerFunction(NameQuantileExactInclusive::name, createAggregateFunctionQuantile<FuncQuantileExactInclusive>);
factory.registerFunction(NameQuantilesExactInclusive::name, createAggregateFunctionQuantile<FuncQuantilesExactInclusive>);
factory.registerFunction(NameQuantileExactWeighted::name, createAggregateFunctionQuantile<FuncQuantileExactWeighted>);
factory.registerFunction(NameQuantilesExactWeighted::name, createAggregateFunctionQuantile<FuncQuantilesExactWeighted>);

9
dbms/src/AggregateFunctions/AggregateFunctionQuantile.h
View File

@ -199,8 +199,15 @@ struct NameQuantileDeterministic { static constexpr auto name = "quantileDetermi
struct NameQuantilesDeterministic { static constexpr auto name = "quantilesDeterministic"; };
struct NameQuantileExact { static constexpr auto name = "quantileExact"; };
struct NameQuantileExactWeighted { static constexpr auto name = "quantileExactWeighted"; };
struct NameQuantilesExact { static constexpr auto name = "quantilesExact"; };
struct NameQuantileExactExclusive { static constexpr auto name = "quantileExactExclusive"; };
struct NameQuantilesExactExclusive { static constexpr auto name = "quantilesExactExclusive"; };
struct NameQuantileExactInclusive { static constexpr auto name = "quantileExactInclusive"; };
struct NameQuantilesExactInclusive { static constexpr auto name = "quantilesExactInclusive"; };
struct NameQuantileExactWeighted { static constexpr auto name = "quantileExactWeighted"; };
struct NameQuantilesExactWeighted { static constexpr auto name = "quantilesExactWeighted"; };
struct NameQuantileTiming { static constexpr auto name = "quantileTiming"; };

4
dbms/src/AggregateFunctions/AggregateFunctionWindowFunnel.cpp
View File

@ -17,8 +17,8 @@ namespace
template <template <typename> class Data>
AggregateFunctionPtr createAggregateFunctionWindowFunnel(const std::string & name, const DataTypes & arguments, const Array & params)
{
if (params.size() != 1)
throw Exception{"Aggregate function " + name + " requires exactly one parameter.", ErrorCodes::NUMBER_OF_ARGUMENTS_DOESNT_MATCH};
if (params.size() < 1)
throw Exception{"Aggregate function " + name + " requires at least one parameter: <window>, [option, [option, ...]]", ErrorCodes::NUMBER_OF_ARGUMENTS_DOESNT_MATCH};
if (arguments.size() < 2)
throw Exception("Aggregate function " + name + " requires one timestamp argument and at least one event condition.", ErrorCodes::NUMBER_OF_ARGUMENTS_DOESNT_MATCH);

16
dbms/src/AggregateFunctions/AggregateFunctionWindowFunnel.h
View File

@ -139,6 +139,7 @@ class AggregateFunctionWindowFunnel final
private:
UInt64 window;
UInt8 events_size;
UInt8 strict;
// Loop through the entire events_list, update the event timestamp value
@ -165,6 +166,10 @@ private:
if (event_idx == 0)
events_timestamp[0] = timestamp;
else if (strict && events_timestamp[event_idx] >= 0)
{
return event_idx + 1;
}
else if (events_timestamp[event_idx - 1] >= 0 && timestamp <= events_timestamp[event_idx - 1] + window)
{
events_timestamp[event_idx] = events_timestamp[event_idx - 1];
@ -191,8 +196,17 @@ public:
{
events_size = arguments.size() - 1;
window = params.at(0).safeGet<UInt64>();
}
strict = 0;
for (size_t i = 1; i < params.size(); ++i)
{
String option = params.at(i).safeGet<String>();
if (option.compare("strict") == 0)
strict = 1;
else
throw Exception{"Aggregate function " + getName() + " doesn't support a parameter: " + option, ErrorCodes::BAD_ARGUMENTS};
}
}
DataTypePtr getReturnType() const override
{

129
dbms/src/AggregateFunctions/QuantileExact.h
View File

@ -14,6 +14,7 @@ namespace DB
namespace ErrorCodes
{
extern const int NOT_IMPLEMENTED;
extern const int BAD_ARGUMENTS;
}
/** Calculates quantile by collecting all values into array
@ -106,16 +107,134 @@ struct QuantileExact
result[i] = Value();
}
}
};
/// The same, but in the case of an empty state, NaN is returned.
Float64 getFloat(Float64) const
/// QuantileExactExclusive is equivalent to Excel PERCENTILE.EXC, R-6, SAS-4, SciPy-(0,0)
template <typename Value>
struct QuantileExactExclusive : public QuantileExact<Value>
{
using QuantileExact<Value>::array;
/// Get the value of the `level` quantile. The level must be between 0 and 1 excluding bounds.
Float64 getFloat(Float64 level)
{
throw Exception("Method getFloat is not implemented for QuantileExact", ErrorCodes::NOT_IMPLEMENTED);
if (!array.empty())
{
if (level == 0. || level == 1.)
throw Exception("QuantileExactExclusive cannot interpolate for the percentiles 1 and 0", ErrorCodes::BAD_ARGUMENTS);
Float64 h = level * (array.size() + 1);
auto n = static_cast<size_t>(h);
if (n >= array.size())
return array[array.size() - 1];
else if (n < 1)
return array[0];
std::nth_element(array.begin(), array.begin() + n - 1, array.end());
auto nth_element = std::min_element(array.begin() + n, array.end());
return array[n - 1] + (h - n) * (*nth_element - array[n - 1]);
}
void getManyFloat(const Float64 *, const size_t *, size_t, Float64 *) const
return std::numeric_limits<Float64>::quiet_NaN();
}
void getManyFloat(const Float64 * levels, const size_t * indices, size_t size, Float64 * result)
{
throw Exception("Method getManyFloat is not implemented for QuantileExact", ErrorCodes::NOT_IMPLEMENTED);
if (!array.empty())
{
size_t prev_n = 0;
for (size_t i = 0; i < size; ++i)
{
auto level = levels[indices[i]];
if (level == 0. || level == 1.)
throw Exception("QuantileExactExclusive cannot interpolate for the percentiles 1 and 0", ErrorCodes::BAD_ARGUMENTS);
Float64 h = level * (array.size() + 1);
auto n = static_cast<size_t>(h);
if (n >= array.size())
result[indices[i]] = array[array.size() - 1];
else if (n < 1)
result[indices[i]] = array[0];
else
{
std::nth_element(array.begin() + prev_n, array.begin() + n - 1, array.end());
auto nth_element = std::min_element(array.begin() + n, array.end());
result[indices[i]] = array[n - 1] + (h - n) * (*nth_element - array[n - 1]);
prev_n = n - 1;
}
}
}
else
{
for (size_t i = 0; i < size; ++i)
result[i] = std::numeric_limits<Float64>::quiet_NaN();
}
}
};
/// QuantileExactInclusive is equivalent to Excel PERCENTILE and PERCENTILE.INC, R-7, SciPy-(1,1)
template <typename Value>
struct QuantileExactInclusive : public QuantileExact<Value>
{
using QuantileExact<Value>::array;
/// Get the value of the `level` quantile. The level must be between 0 and 1 including bounds.
Float64 getFloat(Float64 level)
{
if (!array.empty())
{
Float64 h = level * (array.size() - 1) + 1;
auto n = static_cast<size_t>(h);
if (n >= array.size())
return array[array.size() - 1];
else if (n < 1)
return array[0];
std::nth_element(array.begin(), array.begin() + n - 1, array.end());
auto nth_element = std::min_element(array.begin() + n, array.end());
return array[n - 1] + (h - n) * (*nth_element - array[n - 1]);
}
return std::numeric_limits<Float64>::quiet_NaN();
}
void getManyFloat(const Float64 * levels, const size_t * indices, size_t size, Float64 * result)
{
if (!array.empty())
{
size_t prev_n = 0;
for (size_t i = 0; i < size; ++i)
{
auto level = levels[indices[i]];
Float64 h = level * (array.size() - 1) + 1;
auto n = static_cast<size_t>(h);
if (n >= array.size())
result[indices[i]] = array[array.size() - 1];
else if (n < 1)
result[indices[i]] = array[0];
else
{
std::nth_element(array.begin() + prev_n, array.begin() + n - 1, array.end());
auto nth_element = std::min_element(array.begin() + n, array.end());
result[indices[i]] = array[n - 1] + (h - n) * (*nth_element - array[n - 1]);
prev_n = n - 1;
}
}
}
else
{
for (size_t i = 0; i < size; ++i)
result[i] = std::numeric_limits<Float64>::quiet_NaN();
}
}
};

4
dbms/src/Common/Dwarf.cpp
View File

@ -1,3 +1,5 @@
#ifdef __ELF__
/*
* Copyright 2012-present Facebook, Inc.
*
@ -1031,3 +1033,5 @@ bool Dwarf::LineNumberVM::findAddress(uintptr_t target, Path & file, uint64_t &
}
}
#endif

4
dbms/src/Common/Dwarf.h
View File

@ -1,5 +1,7 @@
#pragma once
#ifdef __ELF__
/*
* Copyright 2012-present Facebook, Inc.
*
@ -285,3 +287,5 @@ private:
};
}
#endif

4
dbms/src/Common/Elf.cpp
View File

@ -1,3 +1,5 @@
#ifdef __ELF__
#include <Common/Elf.h>
#include <Common/Exception.h>
@ -128,3 +130,5 @@ size_t Elf::Section::size() const
}
}
#endif

4
dbms/src/Common/Elf.h
View File

@ -1,5 +1,7 @@
#pragma once
#ifdef __ELF__
#include <IO/MMapReadBufferFromFile.h>
#include <string>
@ -61,3 +63,5 @@ private:
};
}
#endif

6
dbms/src/Common/ErrorCodes.cpp
View File

@ -443,8 +443,10 @@ namespace ErrorCodes
extern const int INSECURE_PATH = 466;
extern const int CANNOT_PARSE_BOOL = 467;
extern const int CANNOT_PTHREAD_ATTR = 468;
extern const int SETTINGS_ARE_NOT_SUPPORTED = 469;
extern const int IMMUTABLE_SETTING = 470;
extern const int QUERY_IS_NOT_SUPPORTED_IN_LIVE_VIEW = 469;
extern const int SETTINGS_ARE_NOT_SUPPORTED = 470;
extern const int IMMUTABLE_SETTING = 471;
extern const int KEEPER_EXCEPTION = 999;
extern const int POCO_EXCEPTION = 1000;

13
dbms/src/Common/StackTrace.cpp
View File

@ -250,6 +250,7 @@ static void toStringEveryLineImpl(const StackTrace::Frames & frames, size_t offs
if (size == 0)
return callback("<Empty trace>");
#ifdef __ELF__
const DB::SymbolIndex & symbol_index = DB::SymbolIndex::instance();
std::unordered_map<std::string, DB::Dwarf> dwarfs;
@ -290,6 +291,18 @@ static void toStringEveryLineImpl(const StackTrace::Frames & frames, size_t offs
callback(out.str());
out.str({});
}
#else
std::stringstream out;
for (size_t i = offset; i < size; ++i)
{
const void * addr = frames[i];
out << i << ". " << addr;
callback(out.str());
out.str({});
}
#endif
}
static std::string toStringImpl(const StackTrace::Frames & frames, size_t offset, size_t size)

4
dbms/src/Common/SymbolIndex.cpp
View File

@ -1,3 +1,5 @@
#ifdef __ELF__
#include <Common/SymbolIndex.h>
#include <algorithm>
@ -316,3 +318,5 @@ const SymbolIndex::Object * SymbolIndex::findObject(const void * address) const
}
}
#endif

4
dbms/src/Common/SymbolIndex.h
View File

@ -1,5 +1,7 @@
#pragma once
#ifdef __ELF__
#include <vector>
#include <string>
#include <ext/singleton.h>
@ -53,3 +55,5 @@ private:
};
}
#endif

10
dbms/src/Common/UInt128.h
View File

@ -1,7 +1,8 @@
#pragma once
#include <tuple>
#include <sstream>
#include <iomanip>
#include <city.h>
#include <Core/Types.h>
@ -33,6 +34,13 @@ struct UInt128
auto tuple() const { return std::tie(high, low); }
String toHexString() const
{
std::ostringstream os;
os << std::setw(16) << std::setfill('0') << std::hex << high << low;
return String(os.str());
}
bool inline operator== (const UInt128 rhs) const { return tuple() == rhs.tuple(); }
bool inline operator!= (const UInt128 rhs) const { return tuple() != rhs.tuple(); }
bool inline operator< (const UInt128 rhs) const { return tuple() < rhs.tuple(); }

7
dbms/src/Common/tests/symbol_index.cpp
View File

@ -16,6 +16,7 @@ using namespace DB;
int main(int argc, char ** argv)
{
#ifdef __ELF__
if (argc < 2)
{
std::cerr << "Usage: ./symbol_index address\n";
@ -53,6 +54,12 @@ int main(int argc, char ** argv)
std::cerr << "\n";
std::cerr << StackTrace().toString() << "\n";
#else
(void)argc;
(void)argv;
std::cerr << "This test does not make sense for non-ELF objects.\n";
#endif
return 0;
}

3
dbms/src/Core/Block.h
View File

@ -144,7 +144,8 @@ private:
using Blocks = std::vector<Block>;
using BlocksList = std::list<Block>;
using BlocksPtr = std::shared_ptr<Blocks>;
using BlocksPtrs = std::shared_ptr<std::vector<BlocksPtr>>;
/// Compare number of columns, data types, column types, column names, and values of constant columns.
bool blocksHaveEqualStructure(const Block & lhs, const Block & rhs);

4
dbms/src/Core/Defines.h
View File

@ -32,7 +32,11 @@
*/
#define DEFAULT_MERGE_BLOCK_SIZE 8192
#define DEFAULT_TEMPORARY_LIVE_VIEW_TIMEOUT_SEC 5
#define DEFAULT_TEMPORARY_LIVE_CHANNEL_TIMEOUT_SEC 15
#define DEFAULT_ALTER_LIVE_CHANNEL_WAIT_MS 10000
#define SHOW_CHARS_ON_SYNTAX_ERROR ptrdiff_t(160)
#define DEFAULT_LIVE_VIEW_HEARTBEAT_INTERVAL_SEC 15
#define DBMS_DEFAULT_DISTRIBUTED_CONNECTIONS_POOL_SIZE 1024
#define DBMS_CONNECTION_POOL_WITH_FAILOVER_DEFAULT_MAX_TRIES 3
/// each period reduces the error counter by 2 times

10
dbms/src/Core/Settings.h
View File

@ -100,7 +100,7 @@ struct Settings : public SettingsCollection<Settings>
M(SettingUInt64, parallel_replicas_count, 0, "") \
M(SettingUInt64, parallel_replica_offset, 0, "") \
\
M(SettingBool, skip_unavailable_shards, false, "Silently skip unavailable shards.") \
M(SettingBool, skip_unavailable_shards, false, "If 1, ClickHouse silently skips unavailable shards and nodes unresolvable through DNS. Shard is marked as unavailable when none of the replicas can be reached.") \
\
M(SettingBool, distributed_group_by_no_merge, false, "Do not merge aggregation states from different servers for distributed query processing - in case it is for certain that there are different keys on different shards.") \
M(SettingBool, optimize_skip_unused_shards, false, "Assumes that data is distributed by sharding_key. Optimization to skip unused shards if SELECT query filters by sharding_key.") \
@ -213,7 +213,7 @@ struct Settings : public SettingsCollection<Settings>
M(SettingUInt64, insert_distributed_timeout, 0, "Timeout for insert query into distributed. Setting is used only with insert_distributed_sync enabled. Zero value means no timeout.") \
M(SettingInt64, distributed_ddl_task_timeout, 180, "Timeout for DDL query responses from all hosts in cluster. If a ddl request has not been performed on all hosts, a response will contain a timeout error and a request will be executed in an async mode. Negative value means infinite.") \
M(SettingMilliseconds, stream_flush_interval_ms, 7500, "Timeout for flushing data from streaming storages.") \
M(SettingMilliseconds, stream_poll_timeout_ms, 500, "Timeout for polling data from streaming storages.") \
M(SettingMilliseconds, stream_poll_timeout_ms, 500, "Timeout for polling data from/to streaming storages.") \
M(SettingString, format_schema, "", "Schema identifier (used by schema-based formats)") \
M(SettingBool, insert_allow_materialized_columns, 0, "If setting is enabled, Allow materialized columns in INSERT.") \
M(SettingSeconds, http_connection_timeout, DEFAULT_HTTP_READ_BUFFER_CONNECTION_TIMEOUT, "HTTP connection timeout.") \
@ -346,6 +346,12 @@ struct Settings : public SettingsCollection<Settings>
/** Obsolete settings that do nothing but left for compatibility reasons. Remove each one after half a year of obsolescence. */ \
\
M(SettingBool, allow_experimental_low_cardinality_type, true, "Obsolete setting, does nothing. Will be removed after 2019-08-13") \
\
M(SettingSeconds, live_view_heartbeat_interval, DEFAULT_LIVE_VIEW_HEARTBEAT_INTERVAL_SEC, "The heartbeat interval in seconds to indicate live query is alive.") \
M(SettingSeconds, temporary_live_view_timeout, DEFAULT_TEMPORARY_LIVE_VIEW_TIMEOUT_SEC, "Timeout after which temporary live view is deleted.") \
M(SettingSeconds, temporary_live_channel_timeout, DEFAULT_TEMPORARY_LIVE_CHANNEL_TIMEOUT_SEC, "Timeout after which temporary live channel is deleted.") \
M(SettingMilliseconds, alter_channel_wait_ms, DEFAULT_ALTER_LIVE_CHANNEL_WAIT_MS, "The wait time for alter channel request.") \
M(SettingUInt64, max_live_view_insert_blocks_before_refresh, 64, "Limit maximum number of inserted blocks after which mergeable blocks are dropped and query is re-executed.") \
DECLARE_SETTINGS_COLLECTION(LIST_OF_SETTINGS)

51
dbms/src/DataStreams/BlocksBlockInputStream.h Normal file
View File

@ -0,0 +1,51 @@
/* Copyright (c) 2018 BlackBerry Limited
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License. */
#pragma once
#include <DataStreams/IBlockInputStream.h>
namespace DB
{
/** A stream of blocks from a shared vector of blocks
*/
class BlocksBlockInputStream : public IBlockInputStream
{
public:
/// Acquires shared ownership of the blocks vector
BlocksBlockInputStream(const std::shared_ptr<BlocksPtr> & blocks_ptr_, Block header_)
: blocks(*blocks_ptr_), it((*blocks_ptr_)->begin()), end((*blocks_ptr_)->end()), header(std::move(header_)) {}
String getName() const override { return "Blocks"; }
Block getHeader() const override { return header; }
protected:
Block readImpl() override
{
if (it == end)
return Block();
Block res = *it;
++it;
return res;
}
private:
BlocksPtr blocks;
Blocks::iterator it;
const Blocks::iterator end;
Block header;
};
}

222
dbms/src/DataStreams/LiveViewBlockInputStream.h Normal file
View File

@ -0,0 +1,222 @@
/* Copyright (c) 2018 BlackBerry Limited
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License. */
#pragma once
#include <limits>
#include <Common/ConcurrentBoundedQueue.h>
#include <Poco/Condition.h>
#include <DataStreams/OneBlockInputStream.h>
#include <DataStreams/IBlockInputStream.h>
#include <Storages/StorageLiveView.h>
namespace DB
{
/** Implements LIVE VIEW table WATCH input stream.
* Keeps stream alive by outputing blocks with no rows
* based on period specified by the heartbeat interval.
*/
class LiveViewBlockInputStream : public IBlockInputStream
{
using NonBlockingResult = std::pair<Block, bool>;
public:
~LiveViewBlockInputStream() override
{
/// Start storage no users thread
/// if we are the last active user
if (!storage->is_dropped && blocks_ptr.use_count() < 3)
storage->startNoUsersThread(temporary_live_view_timeout_sec);
}
LiveViewBlockInputStream(std::shared_ptr<StorageLiveView> storage_,
std::shared_ptr<BlocksPtr> blocks_ptr_,
std::shared_ptr<BlocksMetadataPtr> blocks_metadata_ptr_,
std::shared_ptr<bool> active_ptr_,
const bool has_limit_, const UInt64 limit_,
const UInt64 heartbeat_interval_sec_,
const UInt64 temporary_live_view_timeout_sec_)
: storage(std::move(storage_)), blocks_ptr(std::move(blocks_ptr_)), blocks_metadata_ptr(std::move(blocks_metadata_ptr_)), active_ptr(std::move(active_ptr_)), has_limit(has_limit_), limit(limit_), heartbeat_interval_usec(heartbeat_interval_sec_ * 1000000), temporary_live_view_timeout_sec(temporary_live_view_timeout_sec_)
{
/// grab active pointer
active = active_ptr.lock();
}
String getName() const override { return "LiveViewBlockInputStream"; }
void cancel(bool kill) override
{
if (isCancelled() || storage->is_dropped)
return;
IBlockInputStream::cancel(kill);
Poco::FastMutex::ScopedLock lock(storage->mutex);
storage->condition.broadcast();
}
Block getHeader() const override { return storage->getHeader(); }
void refresh()
{
if (active && blocks && it == end)
it = blocks->begin();
}
void suspend()
{
active.reset();
}
void resume()
{
active = active_ptr.lock();
{
if (!blocks || blocks.get() != (*blocks_ptr).get())
blocks = (*blocks_ptr);
}
it = blocks->begin();
begin = blocks->begin();
end = blocks->end();
}
NonBlockingResult tryRead()
{
return tryRead_(false);
}
protected:
Block readImpl() override
{
/// try reading
return tryRead_(true).first;
}
/** tryRead method attempts to read a block in either blocking
* or non-blocking mode. If blocking is set to false
* then method return empty block with flag set to false
* to indicate that method would block to get the next block.
*/
NonBlockingResult tryRead_(bool blocking)
{
Block res;
if (has_limit && num_updates == static_cast<Int64>(limit))
{
return { Block(), true };
}
/// If blocks were never assigned get blocks
if (!blocks)
{
Poco::FastMutex::ScopedLock lock(storage->mutex);
if (!active)
return { Block(), false };
blocks = (*blocks_ptr);
it = blocks->begin();
begin = blocks->begin();
end = blocks->end();
}
if (isCancelled() || storage->is_dropped)
{
return { Block(), true };
}
if (it == end)
{
{
Poco::FastMutex::ScopedLock lock(storage->mutex);
if (!active)
return { Block(), false };
/// If we are done iterating over our blocks
/// and there are new blocks availble then get them
if (blocks.get() != (*blocks_ptr).get())
{
blocks = (*blocks_ptr);
it = blocks->begin();
begin = blocks->begin();
end = blocks->end();
}
/// No new blocks available wait for new ones
else
{
if (!blocking)
{
return { Block(), false };
}
if (!end_of_blocks)
{
end_of_blocks = true;
return { getHeader(), true };
}
while (true)
{
UInt64 timestamp_usec = static_cast<UInt64>(timestamp.epochMicroseconds());
bool signaled = storage->condition.tryWait(storage->mutex, std::max(static_cast<UInt64>(0), heartbeat_interval_usec - (timestamp_usec - last_event_timestamp_usec)) / 1000);
if (isCancelled() || storage->is_dropped)
{
return { Block(), true };
}
if (signaled)
{
break;
}
else
{
// heartbeat
last_event_timestamp_usec = static_cast<UInt64>(timestamp.epochMicroseconds());
return { getHeader(), true };
}
}
}
}
return tryRead_(blocking);
}
res = *it;
++it;
if (it == end)
{
end_of_blocks = false;
num_updates += 1;
}
last_event_timestamp_usec = static_cast<UInt64>(timestamp.epochMicroseconds());
return { res, true };
}
private:
std::shared_ptr<StorageLiveView> storage;
std::shared_ptr<BlocksPtr> blocks_ptr;
std::shared_ptr<BlocksMetadataPtr> blocks_metadata_ptr;
std::weak_ptr<bool> active_ptr;
std::shared_ptr<bool> active;
BlocksPtr blocks;
BlocksMetadataPtr blocks_metadata;
Blocks::iterator it;
Blocks::iterator end;
Blocks::iterator begin;
const bool has_limit;
const UInt64 limit;
Int64 num_updates = -1;
bool end_of_blocks = false;
UInt64 heartbeat_interval_usec;
UInt64 temporary_live_view_timeout_sec;
UInt64 last_event_timestamp_usec = 0;
Poco::Timestamp timestamp;
};
}

243
dbms/src/DataStreams/LiveViewEventsBlockInputStream.h Normal file
View File

@ -0,0 +1,243 @@
/* Copyright (c) 2018 BlackBerry Limited
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License. */
#pragma once
#include <limits>
#include <Common/ConcurrentBoundedQueue.h>
#include <Poco/Condition.h>
#include <DataTypes/DataTypesNumber.h>
#include <DataTypes/DataTypeString.h>
#include <Columns/ColumnString.h>
#include <Columns/ColumnsNumber.h>
#include <DataStreams/OneBlockInputStream.h>
#include <DataStreams/IBlockInputStream.h>
#include <Storages/StorageLiveView.h>
namespace DB
{
/** Implements LIVE VIEW table WATCH EVENTS input stream.
* Keeps stream alive by outputing blocks with no rows
* based on period specified by the heartbeat interval.
*/
class LiveViewEventsBlockInputStream : public IBlockInputStream
{
using NonBlockingResult = std::pair<Block, bool>;
public:
~LiveViewEventsBlockInputStream() override
{
/// Start storage no users thread
/// if we are the last active user
if (!storage->is_dropped && blocks_ptr.use_count() < 3)
storage->startNoUsersThread(temporary_live_view_timeout_sec);
}
/// length default -2 because we want LIMIT to specify number of updates so that LIMIT 1 waits for 1 update
/// and LIMIT 0 just returns data without waiting for any updates
LiveViewEventsBlockInputStream(std::shared_ptr<StorageLiveView> storage_,
std::shared_ptr<BlocksPtr> blocks_ptr_,
std::shared_ptr<BlocksMetadataPtr> blocks_metadata_ptr_,
std::shared_ptr<bool> active_ptr_,
const bool has_limit_, const UInt64 limit_,
const UInt64 heartbeat_interval_sec_,
const UInt64 temporary_live_view_timeout_sec_)
: storage(std::move(storage_)), blocks_ptr(std::move(blocks_ptr_)), blocks_metadata_ptr(std::move(blocks_metadata_ptr_)), active_ptr(std::move(active_ptr_)), has_limit(has_limit_), limit(limit_), heartbeat_interval_usec(heartbeat_interval_sec_ * 1000000), temporary_live_view_timeout_sec(temporary_live_view_timeout_sec_)
{
/// grab active pointer
active = active_ptr.lock();
}
String getName() const override { return "LiveViewEventsBlockInputStream"; }
void cancel(bool kill) override
{
if (isCancelled() || storage->is_dropped)
return;
IBlockInputStream::cancel(kill);
Poco::FastMutex::ScopedLock lock(storage->mutex);
storage->condition.broadcast();
}
Block getHeader() const override
{
return {ColumnWithTypeAndName(ColumnUInt64::create(), std::make_shared<DataTypeUInt64>(), "version")};
}
void refresh()
{
if (active && blocks && it == end)
it = blocks->begin();
}
void suspend()
{
active.reset();
}
void resume()
{
active = active_ptr.lock();
{
if (!blocks || blocks.get() != (*blocks_ptr).get())
{
blocks = (*blocks_ptr);
blocks_metadata = (*blocks_metadata_ptr);
}
}
it = blocks->begin();
begin = blocks->begin();
end = blocks->end();
}
NonBlockingResult tryRead()
{
return tryRead_(false);
}
Block getEventBlock()
{
Block res{
ColumnWithTypeAndName(
DataTypeUInt64().createColumnConst(1, blocks_metadata->version)->convertToFullColumnIfConst(),
std::make_shared<DataTypeUInt64>(),
"version")
};
return res;
}
protected:
Block readImpl() override
{
/// try reading
return tryRead_(true).first;
}
/** tryRead method attempts to read a block in either blocking
* or non-blocking mode. If blocking is set to false
* then method return empty block with flag set to false
* to indicate that method would block to get the next block.
*/
NonBlockingResult tryRead_(bool blocking)
{
if (has_limit && num_updates == static_cast<Int64>(limit))
{
return { Block(), true };
}
/// If blocks were never assigned get blocks
if (!blocks)
{
Poco::FastMutex::ScopedLock lock(storage->mutex);
if (!active)
return { Block(), false };
blocks = (*blocks_ptr);
blocks_metadata = (*blocks_metadata_ptr);
it = blocks->begin();
begin = blocks->begin();
end = blocks->end();
}
if (isCancelled() || storage->is_dropped)
{
return { Block(), true };
}
if (it == end)
{
{
Poco::FastMutex::ScopedLock lock(storage->mutex);
if (!active)
return { Block(), false };
/// If we are done iterating over our blocks
/// and there are new blocks availble then get them
if (blocks.get() != (*blocks_ptr).get())
{
blocks = (*blocks_ptr);
blocks_metadata = (*blocks_metadata_ptr);
it = blocks->begin();
begin = blocks->begin();
end = blocks->end();
}
/// No new blocks available wait for new ones
else
{
if (!blocking)
{
return { Block(), false };
}
if (!end_of_blocks)
{
end_of_blocks = true;
return { getHeader(), true };
}
while (true)
{
UInt64 timestamp_usec = static_cast<UInt64>(timestamp.epochMicroseconds());
bool signaled = storage->condition.tryWait(storage->mutex, std::max(static_cast<UInt64>(0), heartbeat_interval_usec - (timestamp_usec - last_event_timestamp_usec)) / 1000);
if (isCancelled() || storage->is_dropped)
{
return { Block(), true };
}
if (signaled)
{
break;
}
else
{
// repeat the event block as a heartbeat
last_event_timestamp_usec = static_cast<UInt64>(timestamp.epochMicroseconds());
return { getHeader(), true };
}
}
}
}
return tryRead_(blocking);
}
// move right to the end
it = end;
if (it == end)
{
end_of_blocks = false;
num_updates += 1;
}
last_event_timestamp_usec = static_cast<UInt64>(timestamp.epochMicroseconds());
return { getEventBlock(), true };
}
private:
std::shared_ptr<StorageLiveView> storage;
std::shared_ptr<BlocksPtr> blocks_ptr;
std::shared_ptr<BlocksMetadataPtr> blocks_metadata_ptr;
std::weak_ptr<bool> active_ptr;
std::shared_ptr<bool> active;
BlocksPtr blocks;
BlocksMetadataPtr blocks_metadata;
Blocks::iterator it;
Blocks::iterator end;
Blocks::iterator begin;
const bool has_limit;
const UInt64 limit;
Int64 num_updates = -1;
bool end_of_blocks = false;
UInt64 heartbeat_interval_usec;
UInt64 temporary_live_view_timeout_sec;
UInt64 last_event_timestamp_usec = 0;
Poco::Timestamp timestamp;
};
}

43
dbms/src/DataStreams/PushingToViewsBlockOutputStream.cpp
View File

@ -12,6 +12,7 @@
#include <Common/ThreadPool.h>
#include <Storages/MergeTree/ReplicatedMergeTreeBlockOutputStream.h>
#include <Storages/StorageValues.h>
#include <Storages/StorageLiveView.h>
namespace DB
{
@ -47,17 +48,30 @@ PushingToViewsBlockOutputStream::PushingToViewsBlockOutputStream(
for (const auto & database_table : dependencies)
{
auto dependent_table = context.getTable(database_table.first, database_table.second);
auto & materialized_view = dynamic_cast<const StorageMaterializedView &>(*dependent_table);
StoragePtr inner_table = materialized_view.getTargetTable();
auto query = materialized_view.getInnerQuery();
ASTPtr query;
BlockOutputStreamPtr out;
if (auto * materialized_view = dynamic_cast<const StorageMaterializedView *>(dependent_table.get()))
{
StoragePtr inner_table = materialized_view->getTargetTable();
query = materialized_view->getInnerQuery();
std::unique_ptr<ASTInsertQuery> insert = std::make_unique<ASTInsertQuery>();
insert->database = inner_table->getDatabaseName();
insert->table = inner_table->getTableName();
ASTPtr insert_query_ptr(insert.release());
InterpreterInsertQuery interpreter(insert_query_ptr, *views_context);
BlockIO io = interpreter.execute();
views.emplace_back(ViewInfo{query, database_table.first, database_table.second, io.out});
out = io.out;
}
else if (dynamic_cast<const StorageLiveView *>(dependent_table.get()))
out = std::make_shared<PushingToViewsBlockOutputStream>(
database_table.first, database_table.second, dependent_table, *views_context, ASTPtr(), true);
else
out = std::make_shared<PushingToViewsBlockOutputStream>(
database_table.first, database_table.second, dependent_table, *views_context, ASTPtr());
views.emplace_back(ViewInfo{std::move(query), database_table.first, database_table.second, std::move(out)});
}
}
@ -90,10 +104,18 @@ void PushingToViewsBlockOutputStream::write(const Block & block)
*/
Nested::validateArraySizes(block);
if (auto * live_view = dynamic_cast<StorageLiveView *>(storage.get()))
{
BlockOutputStreamPtr output_ = std::make_shared<LiveViewBlockOutputStream>(*live_view);
StorageLiveView::writeIntoLiveView(*live_view, block, context, output_);
}
else
{
if (output)
/// TODO: to support virtual and alias columns inside MVs, we should return here the inserted block extended
/// with additional columns directly from storage and pass it to MVs instead of raw block.
output->write(block);
}
/// Don't process materialized views if this block is duplicate
if (replicated_output && replicated_output->lastBlockIsDuplicate())
@ -179,21 +201,30 @@ void PushingToViewsBlockOutputStream::process(const Block & block, size_t view_n
auto & view = views[view_num];
try
{
BlockInputStreamPtr in;
if (view.query)
{
/// We create a table with the same name as original table and the same alias columns,
/// but it will contain single block (that is INSERT-ed into main table).
/// InterpreterSelectQuery will do processing of alias columns.
Context local_context = *views_context;
local_context.addViewSource(StorageValues::create(storage->getDatabaseName(), storage->getTableName(), storage->getColumns(), block));
local_context.addViewSource(
StorageValues::create(storage->getDatabaseName(), storage->getTableName(), storage->getColumns(),
block));
InterpreterSelectQuery select(view.query, local_context, SelectQueryOptions());
in = std::make_shared<MaterializingBlockInputStream>(select.execute().in);
BlockInputStreamPtr in = std::make_shared<MaterializingBlockInputStream>(select.execute().in);
/// Squashing is needed here because the materialized view query can generate a lot of blocks
/// even when only one block is inserted into the parent table (e.g. if the query is a GROUP BY
/// and two-level aggregation is triggered).
in = std::make_shared<SquashingBlockInputStream>(
in, context.getSettingsRef().min_insert_block_size_rows, context.getSettingsRef().min_insert_block_size_bytes);
in = std::make_shared<ConvertingBlockInputStream>(context, in, view.out->getHeader(), ConvertingBlockInputStream::MatchColumnsMode::Name);
}
else
in = std::make_shared<OneBlockInputStream>(block);
in->readPrefix();

2
dbms/src/DataStreams/PushingToViewsBlockOutputStream.h
View File

@ -5,7 +5,7 @@
#include <DataStreams/OneBlockInputStream.h>
#include <DataStreams/MaterializingBlockInputStream.h>
#include <Storages/StorageMaterializedView.h>
#include <Storages/StorageLiveView.h>
namespace DB
{

13
dbms/src/DataStreams/RemoteBlockInputStream.cpp
View File

@ -164,7 +164,18 @@ static Block adaptBlockStructure(const Block & block, const Block & header, cons
res.info = block.info;
for (const auto & elem : header)
res.insert({ castColumn(block.getByName(elem.name), elem.type, context), elem.type, elem.name });
{
ColumnPtr column;
if (elem.column && isColumnConst(*elem.column))
/// TODO: check that column from block contains the same value.
/// TODO: serialize const columns.
column = elem.column->cloneResized(block.rows());
else
column = castColumn(block.getByName(elem.name), elem.type, context);
res.insert({column, elem.type, elem.name});
}
return res;
}

2
dbms/src/DataStreams/copyData.cpp
View File

@ -28,6 +28,8 @@ void copyDataImpl(IBlockInputStream & from, IBlockOutputStream & to, TCancelCall
break;
to.write(block);
if (!block.rows())
to.flush();
progress(block);
}

2
dbms/src/Databases/DatabasesCommon.cpp
View File

@ -41,7 +41,7 @@ String getTableDefinitionFromCreateQuery(const ASTPtr & query)
create.replace_view = false;
/// For views it is necessary to save the SELECT query itself, for the rest - on the contrary
if (!create.is_view && !create.is_materialized_view)
if (!create.is_view && !create.is_materialized_view && !create.is_live_view)
create.select = nullptr;
create.format = nullptr;

45
dbms/src/Formats/BlockOutputStreamFromRowOutputStream.cpp
View File

@ -1,45 +0,0 @@
#include <Core/Block.h>
#include <Formats/BlockOutputStreamFromRowOutputStream.h>
namespace DB
{
BlockOutputStreamFromRowOutputStream::BlockOutputStreamFromRowOutputStream(RowOutputStreamPtr row_output_, const Block & header_)
: row_output(row_output_), header(header_) {}
void BlockOutputStreamFromRowOutputStream::write(const Block & block)
{
size_t rows = block.rows();
for (size_t i = 0; i < rows; ++i)
{
if (!first_row)
row_output->writeRowBetweenDelimiter();
first_row = false;
row_output->write(block, i);
}
}
void BlockOutputStreamFromRowOutputStream::setRowsBeforeLimit(size_t rows_before_limit)
{
row_output->setRowsBeforeLimit(rows_before_limit);
}
void BlockOutputStreamFromRowOutputStream::setTotals(const Block & totals)
{
row_output->setTotals(totals);
}
void BlockOutputStreamFromRowOutputStream::setExtremes(const Block & extremes)
{
row_output->setExtremes(extremes);
}
void BlockOutputStreamFromRowOutputStream::onProgress(const Progress & progress)
{
row_output->onProgress(progress);
}
}

38
dbms/src/Formats/BlockOutputStreamFromRowOutputStream.h
View File

@ -1,38 +0,0 @@
#pragma once
#include <DataStreams/IBlockOutputStream.h>
#include <Formats/IRowOutputStream.h>
namespace DB
{
/** Transforms a stream to write data by rows to a stream to write data by blocks.
* For example, to write a text dump.
*/
class BlockOutputStreamFromRowOutputStream : public IBlockOutputStream
{
public:
BlockOutputStreamFromRowOutputStream(RowOutputStreamPtr row_output_, const Block & header_);
Block getHeader() const override { return header; }
void write(const Block & block) override;
void writePrefix() override { row_output->writePrefix(); }
void writeSuffix() override { row_output->writeSuffix(); }
void flush() override { row_output->flush(); }
void setRowsBeforeLimit(size_t rows_before_limit) override;
void setTotals(const Block & totals) override;
void setExtremes(const Block & extremes) override;
void onProgress(const Progress & progress) override;
String getContentType() const override { return row_output->getContentType(); }
private:
RowOutputStreamPtr row_output;
Block header;
bool first_row = true;
};
}

10
dbms/src/Formats/CSVRowInputStream.cpp
View File

@ -348,10 +348,9 @@ bool OPTIMIZE(1) CSVRowInputStream::parseRowAndPrintDiagnosticInfo(MutableColumn
const auto & current_column_type = data_types[table_column];
const bool is_last_file_column =
file_column + 1 == column_indexes_for_input_fields.size();
const bool at_delimiter = *istr.position() == delimiter;
const bool at_delimiter = !istr.eof() && *istr.position() == delimiter;
const bool at_last_column_line_end = is_last_file_column
&& (*istr.position() == '\n' || *istr.position() == '\r'
|| istr.eof());
&& (istr.eof() || *istr.position() == '\n' || *istr.position() == '\r');
out << "Column " << file_column << ", " << std::string((file_column < 10 ? 2 : file_column < 100 ? 1 : 0), ' ')
<< "name: " << header.safeGetByPosition(table_column).name << ", " << std::string(max_length_of_column_name - header.safeGetByPosition(table_column).name.size(), ' ')
@ -514,10 +513,9 @@ void CSVRowInputStream::updateDiagnosticInfo()
bool CSVRowInputStream::readField(IColumn & column, const DataTypePtr & type, bool is_last_file_column, size_t column_idx)
{
const bool at_delimiter = *istr.position() == format_settings.csv.delimiter;
const bool at_delimiter = !istr.eof() || *istr.position() == format_settings.csv.delimiter;
const bool at_last_column_line_end = is_last_file_column
&& (*istr.position() == '\n' || *istr.position() == '\r'
|| istr.eof());
&& (istr.eof() || *istr.position() == '\n' || *istr.position() == '\r');
if (format_settings.csv.empty_as_default
&& (at_delimiter || at_last_column_line_end))

15
dbms/src/Formats/FormatFactory.cpp
View File

@ -100,7 +100,8 @@ BlockInputStreamPtr FormatFactory::getInput(
}
BlockOutputStreamPtr FormatFactory::getOutput(const String & name, WriteBuffer & buf, const Block & sample, const Context & context) const
BlockOutputStreamPtr FormatFactory::getOutput(
const String & name, WriteBuffer & buf, const Block & sample, const Context & context, WriteCallback callback) const
{
if (name == "PrettyCompactMonoBlock")
{
@ -128,10 +129,10 @@ BlockOutputStreamPtr FormatFactory::getOutput(const String & name, WriteBuffer &
* which only work with full columns.
*/
return std::make_shared<MaterializingBlockOutputStream>(
output_getter(buf, sample, context, format_settings), sample);
output_getter(buf, sample, context, callback, format_settings), sample);
}
auto format = getOutputFormat(name, buf, sample, context);
auto format = getOutputFormat(name, buf, sample, context, callback);
return std::make_shared<MaterializingBlockOutputStream>(std::make_shared<OutputStreamToOutputFormat>(format), sample);
}
@ -165,7 +166,8 @@ InputFormatPtr FormatFactory::getInputFormat(
}
OutputFormatPtr FormatFactory::getOutputFormat(const String & name, WriteBuffer & buf, const Block & sample, const Context & context) const
OutputFormatPtr FormatFactory::getOutputFormat(
const String & name, WriteBuffer & buf, const Block & sample, const Context & context, WriteCallback callback) const
{
const auto & output_getter = getCreators(name).output_processor_creator;
if (!output_getter)
@ -177,7 +179,7 @@ OutputFormatPtr FormatFactory::getOutputFormat(const String & name, WriteBuffer
/** TODO: Materialization is needed, because formats can use the functions `IDataType`,
* which only work with full columns.
*/
return output_getter(buf, sample, context, format_settings);
return output_getter(buf, sample, context, callback, format_settings);
}
@ -250,6 +252,7 @@ void registerOutputFormatProcessorPrettySpace(FormatFactory & factory);
void registerOutputFormatProcessorVertical(FormatFactory & factory);
void registerOutputFormatProcessorJSON(FormatFactory & factory);
void registerOutputFormatProcessorJSONCompact(FormatFactory & factory);
void registerOutputFormatProcessorJSONEachRowWithProgress(FormatFactory & factory);
void registerOutputFormatProcessorXML(FormatFactory & factory);
void registerOutputFormatProcessorODBCDriver(FormatFactory & factory);
void registerOutputFormatProcessorODBCDriver2(FormatFactory & factory);
@ -266,6 +269,8 @@ FormatFactory::FormatFactory()
registerInputFormatTabSeparated(*this);
registerInputFormatCSV(*this);
registerOutputFormatProcessorJSONEachRowWithProgress(*this);
registerInputFormatProcessorNative(*this);
registerOutputFormatProcessorNative(*this);
registerInputFormatProcessorRowBinary(*this);

12
dbms/src/Formats/FormatFactory.h
View File

@ -41,6 +41,10 @@ public:
/// It's initial purpose was to extract payload for virtual columns from Kafka Consumer ReadBuffer.
using ReadCallback = std::function<void()>;
/// This callback allows to perform some additional actions after writing a single row.
/// It's initial purpose was to flush Kafka message for each row.
using WriteCallback = std::function<void()>;
private:
using InputCreator = std::function<BlockInputStreamPtr(
ReadBuffer & buf,
@ -55,6 +59,7 @@ private:
WriteBuffer & buf,
const Block & sample,
const Context & context,
WriteCallback callback,
const FormatSettings & settings)>;
using InputProcessorCreator = std::function<InputFormatPtr(
@ -68,6 +73,7 @@ private:
WriteBuffer & buf,
const Block & sample,
const Context & context,
WriteCallback callback,
const FormatSettings & settings)>;
struct Creators
@ -91,7 +97,7 @@ public:
ReadCallback callback = {}) const;
BlockOutputStreamPtr getOutput(const String & name, WriteBuffer & buf,
const Block & sample, const Context & context) const;
const Block & sample, const Context & context, WriteCallback callback = {}) const;
InputFormatPtr getInputFormat(
const String & name,
@ -102,8 +108,8 @@ public:
UInt64 rows_portion_size = 0,
ReadCallback callback = {}) const;
OutputFormatPtr getOutputFormat(const String & name, WriteBuffer & buf,
const Block & sample, const Context & context) const;
OutputFormatPtr getOutputFormat(
const String & name, WriteBuffer & buf, const Block & sample, const Context & context, WriteCallback callback = {}) const;
/// Register format by its name.
void registerInputFormat(const String & name, InputCreator input_creator);

1
dbms/src/Formats/NativeFormat.cpp
View File

@ -27,6 +27,7 @@ void registerOutputFormatNative(FormatFactory & factory)
WriteBuffer & buf,
const Block & sample,
const Context &,
FormatFactory::WriteCallback,
const FormatSettings &)
{
return std::make_shared<NativeBlockOutputStream>(buf, 0, sample);

1
dbms/src/Formats/NullFormat.cpp
View File

@ -11,6 +11,7 @@ void registerOutputFormatNull(FormatFactory & factory)
WriteBuffer &,
const Block & sample,
const Context &,
FormatFactory::WriteCallback,
const FormatSettings &)
{
return std::make_shared<NullBlockOutputStream>(sample);

3
dbms/src/Formats/tests/block_row_transforms.cpp
View File

@ -14,7 +14,6 @@
#include <Formats/TabSeparatedRowInputStream.h>
#include <Formats/BlockInputStreamFromRowInputStream.h>
#include <Formats/BlockOutputStreamFromRowOutputStream.h>
#include <DataStreams/copyData.h>
#include <Processors/Formats/Impl/TabSeparatedRowOutputFormat.h>
@ -47,7 +46,7 @@ try
RowInputStreamPtr row_input = std::make_shared<TabSeparatedRowInputStream>(in_buf, sample, false, false, format_settings);
BlockInputStreamFromRowInputStream block_input(row_input, sample, DEFAULT_INSERT_BLOCK_SIZE, 0, []{}, format_settings);
BlockOutputStreamPtr block_output = std::make_shared<OutputStreamToOutputFormat>(std::make_shared<TabSeparatedRowOutputFormat>(out_buf, sample, false, false, format_settings));
BlockOutputStreamPtr block_output = std::make_shared<OutputStreamToOutputFormat>(std::make_shared<TabSeparatedRowOutputFormat>(out_buf, sample, false, false, []{}, format_settings));
copyData(block_input, *block_output);
}

3
dbms/src/Formats/tests/tab_separated_streams.cpp
View File

@ -11,7 +11,6 @@
#include <Formats/TabSeparatedRowInputStream.h>
#include <Formats/BlockInputStreamFromRowInputStream.h>
#include <Formats/BlockOutputStreamFromRowOutputStream.h>
#include <DataStreams/copyData.h>
#include <Processors/Formats/OutputStreamToOutputFormat.h>
@ -44,7 +43,7 @@ try
BlockInputStreamFromRowInputStream block_input(row_input, sample, DEFAULT_INSERT_BLOCK_SIZE, 0, []{}, format_settings);
BlockOutputStreamPtr block_output = std::make_shared<OutputStreamToOutputFormat>(
std::make_shared<TabSeparatedRowOutputFormat>(out_buf, sample, false, false, format_settings));
std::make_shared<TabSeparatedRowOutputFormat>(out_buf, sample, false, false, [] {}, format_settings));
copyData(block_input, *block_output);
return 0;

11
dbms/src/Functions/IFunction.h
View File

@ -159,6 +159,13 @@ public:
*/
virtual bool isSuitableForConstantFolding() const { return true; }
/** Some functions like ignore(...) or toTypeName(...) always return constant result which doesn't depend on arguments.
* In this case we can calculate result and assume that it's constant in stream header.
* There is no need to implement function if it has zero arguments.
* Must return ColumnConst with single row or nullptr.
*/
virtual ColumnPtr getResultIfAlwaysReturnsConstantAndHasArguments(const Block & /*block*/, const ColumnNumbers & /*arguments*/) const { return nullptr; }
/** Function is called "injective" if it returns different result for different values of arguments.
* Example: hex, negate, tuple...
*
@ -456,6 +463,10 @@ public:
}
bool isSuitableForConstantFolding() const override { return function->isSuitableForConstantFolding(); }
ColumnPtr getResultIfAlwaysReturnsConstantAndHasArguments(const Block & block, const ColumnNumbers & arguments_) const override
{
return function->getResultIfAlwaysReturnsConstantAndHasArguments(block, arguments_);
}
bool isInjective(const Block & sample_block) override { return function->isInjective(sample_block); }

4
dbms/src/Functions/addressToLine.cpp
View File

@ -1,3 +1,5 @@
#ifdef __ELF__
#include <Common/Elf.h>
#include <Common/Dwarf.h>
#include <Common/SymbolIndex.h>
@ -149,3 +151,5 @@ void registerFunctionAddressToLine(FunctionFactory & factory)
}
}
#endif

4
dbms/src/Functions/addressToSymbol.cpp
View File

@ -1,3 +1,5 @@
#ifdef __ELF__
#include <Common/SymbolIndex.h>
#include <Columns/ColumnString.h>
#include <Columns/ColumnsNumber.h>
@ -92,3 +94,5 @@ void registerFunctionAddressToSymbol(FunctionFactory & factory)
}
}
#endif

2
dbms/src/Functions/array/arrayWithConstant.cpp
View File

@ -72,7 +72,7 @@ public:
offsets.push_back(offset);
}
block.getByPosition(result).column = ColumnArray::create(col_value->replicate(offsets), std::move(offsets_col));
block.getByPosition(result).column = ColumnArray::create(col_value->replicate(offsets)->convertToFullColumnIfConst(), std::move(offsets_col));
}
};

6
dbms/src/Functions/defaultValueOfArgumentType.cpp
View File

@ -37,6 +37,12 @@ public:
const IDataType & type = *block.getByPosition(arguments[0]).type;
block.getByPosition(result).column = type.createColumnConst(input_rows_count, type.getDefault());
}
ColumnPtr getResultIfAlwaysReturnsConstantAndHasArguments(const Block & block, const ColumnNumbers & arguments) const override
{
const IDataType & type = *block.getByPosition(arguments[0]).type;
return type.createColumnConst(1, type.getDefault());
}
};

9
dbms/src/Functions/getSizeOfEnumType.cpp
View File

@ -49,11 +49,16 @@ public:
}
void executeImpl(Block & block, const ColumnNumbers & arguments, size_t result, size_t input_rows_count) override
{
block.getByPosition(result).column = getResultIfAlwaysReturnsConstantAndHasArguments(block, arguments)->cloneResized(input_rows_count);
}
ColumnPtr getResultIfAlwaysReturnsConstantAndHasArguments(const Block & block, const ColumnNumbers & arguments) const override
{
if (auto type8 = checkAndGetDataType<DataTypeEnum8>(block.getByPosition(arguments[0]).type.get()))
block.getByPosition(result).column = DataTypeUInt8().createColumnConst(input_rows_count, type8->getValues().size());
return DataTypeUInt8().createColumnConst(1, type8->getValues().size());
else if (auto type16 = checkAndGetDataType<DataTypeEnum16>(block.getByPosition(arguments[0]).type.get()))
block.getByPosition(result).column = DataTypeUInt16().createColumnConst(input_rows_count, type16->getValues().size());
return DataTypeUInt16().createColumnConst(1, type16->getValues().size());
else
throw Exception("The argument for function " + getName() + " must be Enum", ErrorCodes::ILLEGAL_TYPE_OF_ARGUMENT);
}

5
dbms/src/Functions/ignore.cpp
View File

@ -42,6 +42,11 @@ public:
{
block.getByPosition(result).column = DataTypeUInt8().createColumnConst(input_rows_count, 0u);
}
ColumnPtr getResultIfAlwaysReturnsConstantAndHasArguments(const Block &, const ColumnNumbers &) const override
{
return DataTypeUInt8().createColumnConst(1, 0u);
}
};

7
dbms/src/Functions/ignoreExceptNull.cpp
View File

@ -1,6 +1,7 @@
#include <Functions/IFunction.h>
#include <Functions/FunctionFactory.h>
#include <DataTypes/DataTypesNumber.h>
#include <Columns/ColumnsNumber.h>
namespace DB
@ -38,7 +39,11 @@ namespace DB
void executeImpl(Block & block, const ColumnNumbers &, size_t result, size_t input_rows_count) override
{
block.getByPosition(result).column = DataTypeUInt8().createColumnConst(input_rows_count, UInt64(0));
/// This function is mainly used in query analysis instead of "in" functions
/// in the case when only header is needed and set for in is not calculated.
/// Because of that function must return the same column type as "in" function, which is ColumnUInt8.
auto res = ColumnUInt8::create(input_rows_count, 0);
block.getByPosition(result).column = std::move(res);
}
};

2
dbms/src/Functions/in.cpp
View File

@ -75,6 +75,8 @@ public:
void executeImpl(Block & block, const ColumnNumbers & arguments, size_t result, size_t /*input_rows_count*/) override
{
/// NOTE: after updating this code, check that FunctionIgnoreExceptNull returns the same type of column.
/// Second argument must be ColumnSet.
ColumnPtr column_set_ptr = block.getByPosition(arguments[1]).column;
const ColumnSet * column_set = typeid_cast<const ColumnSet *>(&*column_set_ptr);

22
dbms/src/Functions/nullIf.cpp
View File

@ -43,18 +43,17 @@ public:
void executeImpl(Block & block, const ColumnNumbers & arguments, size_t result, size_t input_rows_count) override
{
/// nullIf(col1, col2) == if(col1 != col2, col1, NULL)
/// nullIf(col1, col2) == if(col1 = col2, NULL, 1)
Block temp_block = block;
size_t res_pos = temp_block.columns();
temp_block.insert({nullptr, std::make_shared<DataTypeUInt8>(), ""});
{
auto equals_func = FunctionFactory::instance().get("notEquals", context)->build(
auto equals_func = FunctionFactory::instance().get("equals", context)->build(
{temp_block.getByPosition(arguments[0]), temp_block.getByPosition(arguments[1])});
equals_func->execute(temp_block, {arguments[0], arguments[1]}, res_pos, input_rows_count);
}
size_t equals_res_pos = temp_block.columns();
temp_block.insert({nullptr, equals_func->getReturnType(), ""});
equals_func->execute(temp_block, {arguments[0], arguments[1]}, equals_res_pos, input_rows_count);
/// Argument corresponding to the NULL value.
size_t null_pos = temp_block.columns();
@ -68,15 +67,14 @@ public:
temp_block.insert(null_elem);
auto func_if = FunctionFactory::instance().get("if", context)->build(
{temp_block.getByPosition(res_pos), temp_block.getByPosition(null_pos), temp_block.getByPosition(arguments[0])});
func_if->execute(temp_block, {res_pos, arguments[0], null_pos}, result, input_rows_count);
{temp_block.getByPosition(equals_res_pos), temp_block.getByPosition(null_pos), temp_block.getByPosition(arguments[0])});
func_if->execute(temp_block, {equals_res_pos, null_pos, arguments[0]}, result, input_rows_count);
block.getByPosition(result).column = std::move(temp_block.getByPosition(result).column);
block.getByPosition(result).column = makeNullable(std::move(temp_block.getByPosition(result).column));
}
};
void registerFunctionNullIf(FunctionFactory & factory)
{
factory.registerFunction<FunctionNullIf>(FunctionFactory::CaseInsensitive);

5
dbms/src/Functions/toColumnTypeName.cpp
View File

@ -38,6 +38,11 @@ public:
block.getByPosition(result).column
= DataTypeString().createColumnConst(input_rows_count, block.getByPosition(arguments[0]).column->getName());
}
ColumnPtr getResultIfAlwaysReturnsConstantAndHasArguments(const Block & block, const ColumnNumbers & arguments) const override
{
return DataTypeString().createColumnConst(1, block.getByPosition(arguments[0]).type->createColumn()->getName());
}
};

80
dbms/src/Functions/toTypeName.cpp
View File

@ -9,33 +9,16 @@ namespace DB
/** toTypeName(x) - get the type name
* Returns name of IDataType instance (name of data type).
*/
class FunctionToTypeName : public IFunction
class PreparedFunctionToTypeName : public PreparedFunctionImpl
{
public:
static constexpr auto name = "toTypeName";
static FunctionPtr create(const Context &)
{
return std::make_shared<FunctionToTypeName>();
}
String getName() const override
{
return name;
}
String getName() const override { return name; }
protected:
bool useDefaultImplementationForNulls() const override { return false; }
bool useDefaultImplementationForLowCardinalityColumns() const override { return false; }
size_t getNumberOfArguments() const override
{
return 1;
}
DataTypePtr getReturnTypeImpl(const DataTypes & /*arguments*/) const override
{
return std::make_shared<DataTypeString>();
}
/// Execute the function on the block.
void executeImpl(Block & block, const ColumnNumbers & arguments, size_t result, size_t input_rows_count) override
{
@ -45,9 +28,64 @@ public:
};
class BaseFunctionToTypeName : public IFunctionBase
{
public:
BaseFunctionToTypeName(DataTypes argument_types_, DataTypePtr return_type_)
: argument_types(std::move(argument_types_)), return_type(std::move(return_type_)) {}
static constexpr auto name = "toTypeName";
String getName() const override { return name; }
const DataTypes & getArgumentTypes() const override { return argument_types; }
const DataTypePtr & getReturnType() const override { return return_type; }
PreparedFunctionPtr prepare(const Block &, const ColumnNumbers &, size_t) const override
{
return std::make_shared<PreparedFunctionToTypeName>();
}
ColumnPtr getResultIfAlwaysReturnsConstantAndHasArguments(const Block &, const ColumnNumbers &) const override
{
return DataTypeString().createColumnConst(1, argument_types.at(0)->getName());
}
private:
DataTypes argument_types;
DataTypePtr return_type;
};
class FunctionToTypeNameBuilder : public FunctionBuilderImpl
{
public:
static constexpr auto name = "toTypeName";
String getName() const override { return name; }
static FunctionBuilderPtr create(const Context &) { return std::make_shared<FunctionToTypeNameBuilder>(); }
size_t getNumberOfArguments() const override { return 1; }
protected:
DataTypePtr getReturnTypeImpl(const DataTypes &) const override { return std::make_shared<DataTypeString>(); }
FunctionBasePtr buildImpl(const ColumnsWithTypeAndName & arguments, const DataTypePtr & return_type) const override
{
DataTypes types;
types.reserve(arguments.size());
for (auto & elem : arguments)
types.emplace_back(elem.type);
return std::make_shared<BaseFunctionToTypeName>(types, return_type);
}
bool useDefaultImplementationForNulls() const override { return false; }
bool useDefaultImplementationForLowCardinalityColumns() const override { return false; }
};
void registerFunctionToTypeName(FunctionFactory & factory)
{
factory.registerFunction<FunctionToTypeName>();
factory.registerFunction<FunctionToTypeNameBuilder>();
}
}

2
dbms/src/IO/MemoryReadWriteBuffer.h
View File

@ -11,7 +11,7 @@
namespace DB
{
/// Stores data in memory chunks, size of cunks are exponentially increasing during write
/// Stores data in memory chunks, size of chunks are exponentially increasing during write
/// Written data could be reread after write
class MemoryWriteBuffer : public WriteBuffer, public IReadableWriteBuffer, boost::noncopyable, private Allocator<false>
{

2
dbms/src/IO/ReadBufferFromFile.cpp
View File

@ -47,7 +47,7 @@ ReadBufferFromFile::ReadBufferFromFile(
if (o_direct)
{
if (fcntl(fd, F_NOCACHE, 1) == -1)
throwFromErrno("Cannot set F_NOCACHE on file " + file_name, file_name, ErrorCodes::CANNOT_OPEN_FILE);
throwFromErrnoWithPath("Cannot set F_NOCACHE on file " + file_name, file_name, ErrorCodes::CANNOT_OPEN_FILE);
}
#endif
}

2
dbms/src/IO/WriteBuffer.h
View File

@ -35,7 +35,7 @@ public:
*/
inline void next()
{
if (!offset())
if (!offset() && available())
return;
bytes += offset();

2
dbms/src/IO/WriteBufferFromFile.cpp
View File

@ -51,7 +51,7 @@ WriteBufferFromFile::WriteBufferFromFile(
if (o_direct)
{
if (fcntl(fd, F_NOCACHE, 1) == -1)
throwFromErrno("Cannot set F_NOCACHE on file " + file_name, file_name, ErrorCodes::CANNOT_OPEN_FILE);
throwFromErrnoWithPath("Cannot set F_NOCACHE on file " + file_name, file_name, ErrorCodes::CANNOT_OPEN_FILE);
}
#endif
}

3
dbms/src/IO/WriteBufferValidUTF8.cpp
View File

@ -117,6 +117,9 @@ void WriteBufferValidUTF8::nextImpl()
memory[i] = p[i];
working_buffer = Buffer(&memory[cnt], memory.data() + memory.size());
/// Propagate next() to the output buffer
output_buffer.next();
}

8
dbms/src/Interpreters/ClusterProxy/SelectStreamFactory.cpp
View File

@ -68,7 +68,13 @@ BlockInputStreamPtr createLocalStream(const ASTPtr & query_ast, const Context &
* If you do not do this, different types (Const and non-Const) columns will be produced in different threads,
* And this is not allowed, since all code is based on the assumption that in the block stream all types are the same.
*/
return std::make_shared<MaterializingBlockInputStream>(stream);
/* Now we don't need to materialize constants, because RemoteBlockInputStream will ignore constant and take it from header.
* So, streams from different threads will always have the same header.
*/
/// return std::make_shared<MaterializingBlockInputStream>(stream);
return stream;
}
}

76
dbms/src/Interpreters/ExpressionActions.cpp
View File

@ -12,6 +12,7 @@
#include <Functions/IFunction.h>
#include <set>
#include <optional>
#include <DataTypes/DataTypeNullable.h>
namespace ProfileEvents
@ -159,20 +160,24 @@ ExpressionAction ExpressionAction::arrayJoin(const NameSet & array_joined_column
}
ExpressionAction ExpressionAction::ordinaryJoin(
const ASTTableJoin & join_params,
std::shared_ptr<const Join> join_,
const Names & join_key_names_left,
const Names & join_key_names_right,
const NamesAndTypesList & columns_added_by_join_)
{
ExpressionAction a;
a.type = JOIN;
a.join = std::move(join_);
a.join_kind = join_params.kind;
a.join_key_names_left = join_key_names_left;
a.join_key_names_right = join_key_names_right;
a.columns_added_by_join = columns_added_by_join_;
return a;
}
void ExpressionAction::prepare(Block & sample_block, const Settings & settings)
void ExpressionAction::prepare(Block & sample_block, const Settings & settings, NameSet & names_not_for_constant_folding)
{
// std::cerr << "preparing: " << toString() << std::endl;
@ -187,6 +192,7 @@ void ExpressionAction::prepare(Block & sample_block, const Settings & settings)
throw Exception("Column '" + result_name + "' already exists", ErrorCodes::DUPLICATE_COLUMN);
bool all_const = true;
bool all_suitable_for_constant_folding = true;
ColumnNumbers arguments(argument_names.size());
for (size_t i = 0; i < argument_names.size(); ++i)
@ -195,6 +201,9 @@ void ExpressionAction::prepare(Block & sample_block, const Settings & settings)
ColumnPtr col = sample_block.safeGetByPosition(arguments[i]).column;
if (!col || !isColumnConst(*col))
all_const = false;
if (names_not_for_constant_folding.count(argument_names[i]))
all_suitable_for_constant_folding = false;
}
size_t result_position = sample_block.columns();
@ -229,6 +238,22 @@ void ExpressionAction::prepare(Block & sample_block, const Settings & settings)
if (col.column->empty())
col.column = col.column->cloneResized(1);
if (!all_suitable_for_constant_folding)
names_not_for_constant_folding.insert(result_name);
}
}
/// Some functions like ignore() or getTypeName() always return constant result even if arguments are not constant.
/// We can't do constant folding, but can specify in sample block that function result is constant to avoid
/// unnecessary materialization.
auto & res = sample_block.getByPosition(result_position);
if (!res.column && function_base->isSuitableForConstantFolding())
{
if (auto col = function_base->getResultIfAlwaysReturnsConstantAndHasArguments(sample_block, arguments))
{
res.column = std::move(col);
names_not_for_constant_folding.insert(result_name);
}
}
@ -252,10 +277,50 @@ void ExpressionAction::prepare(Block & sample_block, const Settings & settings)
case JOIN:
{
/// TODO join_use_nulls setting
bool is_null_used_as_default = settings.join_use_nulls;
bool right_or_full_join = join_kind == ASTTableJoin::Kind::Right || join_kind == ASTTableJoin::Kind::Full;
bool left_or_full_join = join_kind == ASTTableJoin::Kind::Left || join_kind == ASTTableJoin::Kind::Full;
for (auto & col : sample_block)
{
/// Materialize column.
/// Column is not empty if it is constant, but after Join all constants will be materialized.
/// So, we need remove constants from header.
if (col.column)
col.column = nullptr;
bool make_nullable = is_null_used_as_default && right_or_full_join;
if (make_nullable && !col.type->isNullable())
col.type = std::make_shared<DataTypeNullable>(col.type);
}
for (const auto & col : columns_added_by_join)
sample_block.insert(ColumnWithTypeAndName(nullptr, col.type, col.name));
{
auto res_type = col.type;
bool make_nullable = is_null_used_as_default && left_or_full_join;
if (!make_nullable)
{
/// Keys from right table are usually not stored in Join, but copied from the left one.
/// So, if left key is nullable, let's make right key nullable too.
/// Note: for some join types it's not needed and, probably, may be removed.
/// Note: changing this code, take into account the implementation in Join.cpp.
auto it = std::find(join_key_names_right.begin(), join_key_names_right.end(), col.name);
if (it != join_key_names_right.end())
{
auto pos = it - join_key_names_right.begin();
const auto & left_key_name = join_key_names_left[pos];
make_nullable = sample_block.getByName(left_key_name).type->isNullable();
}
}
if (make_nullable && !res_type->isNullable())
res_type = std::make_shared<DataTypeNullable>(res_type);
sample_block.insert(ColumnWithTypeAndName(nullptr, res_type, col.name));
}
break;
}
@ -683,7 +748,7 @@ void ExpressionActions::addImpl(ExpressionAction action, Names & new_names)
for (const auto & name_with_alias : action.projection)
new_names.emplace_back(name_with_alias.second);
action.prepare(sample_block, settings);
action.prepare(sample_block, settings, names_not_for_constant_folding);
actions.push_back(action);
}
@ -915,7 +980,7 @@ void ExpressionActions::finalize(const Names & output_columns)
if (action.type == ExpressionAction::APPLY_FUNCTION && sample_block.has(out))
{
auto & result = sample_block.getByName(out);
if (result.column)
if (result.column && names_not_for_constant_folding.count(result.name) == 0)
{
action.type = ExpressionAction::ADD_COLUMN;
action.result_type = result.type;
@ -1262,6 +1327,7 @@ bool ExpressionAction::operator==(const ExpressionAction & other) const
&& array_join_is_left == other.array_join_is_left
&& join == other.join
&& join_key_names_left == other.join_key_names_left
&& join_key_names_right == other.join_key_names_right
&& columns_added_by_join == other.columns_added_by_join
&& projection == other.projection
&& is_function_compiled == other.is_function_compiled;

10
dbms/src/Interpreters/ExpressionActions.h
View File

@ -10,6 +10,7 @@
#include "config_core.h"
#include <unordered_map>
#include <unordered_set>
#include <Parsers/ASTTablesInSelectQuery.h>
namespace DB
@ -104,7 +105,9 @@ public:
/// For JOIN
std::shared_ptr<const Join> join;
ASTTableJoin::Kind join_kind;
Names join_key_names_left;
Names join_key_names_right;
NamesAndTypesList columns_added_by_join;
/// For PROJECT.
@ -121,7 +124,8 @@ public:
static ExpressionAction project(const Names & projected_columns_);
static ExpressionAction addAliases(const NamesWithAliases & aliased_columns_);
static ExpressionAction arrayJoin(const NameSet & array_joined_columns, bool array_join_is_left, const Context & context);
static ExpressionAction ordinaryJoin(std::shared_ptr<const Join> join_, const Names & join_key_names_left,
static ExpressionAction ordinaryJoin(const ASTTableJoin & join_params, std::shared_ptr<const Join> join_,
const Names & join_key_names_left, const Names & join_key_names_right,
const NamesAndTypesList & columns_added_by_join_);
/// Which columns necessary to perform this action.
@ -139,7 +143,7 @@ public:
private:
friend class ExpressionActions;
void prepare(Block & sample_block, const Settings & settings);
void prepare(Block & sample_block, const Settings & settings, NameSet & names_not_for_constant_folding);
void execute(Block & block, bool dry_run) const;
void executeOnTotals(Block & block) const;
};
@ -263,6 +267,8 @@ private:
Actions actions;
/// The example of result (output) block.
Block sample_block;
/// Columns which can't be used for constant folding.
NameSet names_not_for_constant_folding;
Settings settings;
#if USE_EMBEDDED_COMPILER

10
dbms/src/Interpreters/ExpressionAnalyzer.cpp
View File

@ -140,7 +140,7 @@ void ExpressionAnalyzer::analyzeAggregation()
for (const auto & key_ast : analyzedJoin().key_asts_left)
getRootActions(key_ast, true, temp_actions);
addJoinAction(temp_actions);
addJoinAction(table_join, temp_actions);
}
}
@ -424,9 +424,9 @@ static void appendRequiredColumns(
}
/// It's possible to set nullptr as join for only_types mode
void ExpressionAnalyzer::addJoinAction(ExpressionActionsPtr & actions, JoinPtr join) const
void ExpressionAnalyzer::addJoinAction(const ASTTableJoin & join_params, ExpressionActionsPtr & actions, JoinPtr join) const
{
actions->add(ExpressionAction::ordinaryJoin(join, analyzedJoin().key_names_left, columnsAddedByJoin()));
actions->add(ExpressionAction::ordinaryJoin(join_params, std::move(join), analyzedJoin().key_names_left, analyzedJoin().key_names_right, columnsAddedByJoin()));
}
bool SelectQueryExpressionAnalyzer::appendJoin(ExpressionActionsChain & chain, bool only_types)
@ -443,8 +443,10 @@ bool SelectQueryExpressionAnalyzer::appendJoin(ExpressionActionsChain & chain, b
initChain(chain, sourceColumns());
ExpressionActionsChain::Step & step = chain.steps.back();
auto & join_params = ast_join->table_join->as<ASTTableJoin &>();
getRootActions(left_keys_list, only_types, step.actions);
addJoinAction(step.actions, subquery_for_set.join);
addJoinAction(join_params, step.actions, subquery_for_set.join);
return true;
}

2
dbms/src/Interpreters/ExpressionAnalyzer.h
View File

@ -131,7 +131,7 @@ protected:
void addMultipleArrayJoinAction(ExpressionActionsPtr & actions, bool is_left) const;
void addJoinAction(ExpressionActionsPtr & actions, JoinPtr join = {}) const;
void addJoinAction(const ASTTableJoin & join_params, ExpressionActionsPtr & actions, JoinPtr join = {}) const;
void getRootActions(const ASTPtr & ast, bool no_subqueries, ExpressionActionsPtr & actions, bool only_consts = false);

7
dbms/src/Interpreters/ExternalLoader.cpp
View File

@ -400,6 +400,10 @@ public:
return count;
}
#if !__clang__
#pragma GCC diagnostic push
#pragma GCC diagnostic ignored "-Wunused-variable"
#endif
bool hasCurrentlyLoadedObjects() const
{
std::lock_guard lock{mutex};
@ -408,6 +412,9 @@ public:
return true;
return false;
}
#if !__clang__
#pragma GCC diagnostic pop
#endif
/// Starts loading of a specified object.
void load(const String & name)

23
dbms/src/Interpreters/InterpreterAlterQuery.cpp
View File

@ -8,7 +8,9 @@
#include <Storages/AlterCommands.h>
#include <Storages/MutationCommands.h>
#include <Storages/PartitionCommands.h>
#include <Storages/LiveViewCommands.h>
#include <Common/typeid_cast.h>
#include <Storages/StorageLiveView.h>
#include <algorithm>
@ -48,6 +50,7 @@ BlockIO InterpreterAlterQuery::execute()
AlterCommands alter_commands;
PartitionCommands partition_commands;
MutationCommands mutation_commands;
LiveViewCommands live_view_commands;
for (ASTAlterCommand * command_ast : alter.command_list->commands)
{
if (auto alter_command = AlterCommand::parse(command_ast))
@ -56,13 +59,16 @@ BlockIO InterpreterAlterQuery::execute()
partition_commands.emplace_back(std::move(*partition_command));
else if (auto mut_command = MutationCommand::parse(command_ast))
mutation_commands.emplace_back(std::move(*mut_command));
else if (auto live_view_command = LiveViewCommand::parse(command_ast))
live_view_commands.emplace_back(std::move(*live_view_command));
else
throw Exception("Wrong parameter type in ALTER query", ErrorCodes::LOGICAL_ERROR);
}
if (!mutation_commands.empty())
{
MutationsInterpreter(table, mutation_commands, context).validate();
auto table_lock_holder = table->lockStructureForShare(false /* because mutation is executed asyncronously */, context.getCurrentQueryId());
MutationsInterpreter(table, mutation_commands, context).validate(table_lock_holder);
table->mutate(mutation_commands, context);
}
@ -72,6 +78,21 @@ BlockIO InterpreterAlterQuery::execute()
table->alterPartition(query_ptr, partition_commands, context);
}
if (!live_view_commands.empty())
{
live_view_commands.validate(*table);
for (const LiveViewCommand & command : live_view_commands)
{
auto live_view = std::dynamic_pointer_cast<StorageLiveView>(table);
switch (command.type)
{
case LiveViewCommand::REFRESH:
live_view->refresh(context);
break;
}
}
}
if (!alter_commands.empty())
{
auto table_lock_holder = table->lockAlterIntention(context.getCurrentQueryId());

17
dbms/src/Interpreters/InterpreterCreateQuery.cpp
View File

@ -442,7 +442,7 @@ void InterpreterCreateQuery::setEngine(ASTCreateQuery & create) const
return;
}
if (create.temporary)
if (create.temporary && !create.is_live_view)
{
auto engine_ast = std::make_shared<ASTFunction>();
engine_ast->name = "Memory";
@ -465,6 +465,11 @@ void InterpreterCreateQuery::setEngine(ASTCreateQuery & create) const
"Cannot CREATE a table AS " + as_database_name + "." + as_table_name + ", it is a View",
ErrorCodes::INCORRECT_QUERY);
if (as_create.is_live_view)
throw Exception(
"Cannot CREATE a table AS " + as_database_name + "." + as_table_name + ", it is a Live View",
ErrorCodes::INCORRECT_QUERY);
create.set(create.storage, as_create.storage->ptr());
}
}
@ -482,7 +487,7 @@ BlockIO InterpreterCreateQuery::createTable(ASTCreateQuery & create)
}
/// Temporary tables are created out of databases.
if (create.temporary && !create.database.empty())
if (create.temporary && !create.database.empty() && !create.is_live_view)
throw Exception("Temporary tables cannot be inside a database. You should not specify a database for a temporary table.",
ErrorCodes::BAD_DATABASE_FOR_TEMPORARY_TABLE);
@ -505,7 +510,7 @@ BlockIO InterpreterCreateQuery::createTable(ASTCreateQuery & create)
if (create.to_database.empty())
create.to_database = current_database;
if (create.select && (create.is_view || create.is_materialized_view))
if (create.select && (create.is_view || create.is_materialized_view || create.is_live_view))
{
AddDefaultDatabaseVisitor visitor(current_database);
visitor.visit(*create.select);
@ -565,7 +570,7 @@ BlockIO InterpreterCreateQuery::createTable(ASTCreateQuery & create)
String data_path;
DatabasePtr database;
if (!create.temporary)
if (!create.temporary || create.is_live_view)
{
database = context.getDatabase(database_name);
data_path = database->getDataPath();
@ -611,7 +616,7 @@ BlockIO InterpreterCreateQuery::createTable(ASTCreateQuery & create)
false);
}
if (create.temporary)
if (create.temporary && !create.is_live_view)
context.getSessionContext().addExternalTable(table_name, res, query_ptr);
else
database->createTable(context, table_name, res, query_ptr);
@ -630,7 +635,7 @@ BlockIO InterpreterCreateQuery::createTable(ASTCreateQuery & create)
/// If the query is a CREATE SELECT, insert the data into the table.
if (create.select && !create.attach
&& !create.is_view && (!create.is_materialized_view || create.is_populate))
&& !create.is_view && !create.is_live_view && (!create.is_materialized_view || create.is_populate))
{
auto insert = std::make_shared<ASTInsertQuery>();

6
dbms/src/Interpreters/InterpreterFactory.cpp
View File

@ -14,6 +14,7 @@
#include <Parsers/ASTUseQuery.h>
#include <Parsers/ASTExplainQuery.h>
#include <Parsers/TablePropertiesQueriesASTs.h>
#include <Parsers/ASTWatchQuery.h>
#include <Interpreters/InterpreterAlterQuery.h>
#include <Interpreters/InterpreterCheckQuery.h>
@ -35,6 +36,7 @@
#include <Interpreters/InterpreterShowTablesQuery.h>
#include <Interpreters/InterpreterSystemQuery.h>
#include <Interpreters/InterpreterUseQuery.h>
#include <Interpreters/InterpreterWatchQuery.h>
#include <Parsers/ASTSystemQuery.h>
@ -173,6 +175,10 @@ std::unique_ptr<IInterpreter> InterpreterFactory::get(ASTPtr & query, Context &
throwIfNoAccess(context);
return std::make_unique<InterpreterSystemQuery>(query, context);
}
else if (query->as<ASTWatchQuery>())
{
return std::make_unique<InterpreterWatchQuery>(query, context);
}
else
throw Exception("Unknown type of query: " + query->getID(), ErrorCodes::UNKNOWN_TYPE_OF_QUERY);
}

1
dbms/src/Interpreters/InterpreterOptimizeQuery.cpp
View File

@ -23,7 +23,6 @@ BlockIO InterpreterOptimizeQuery::execute()
return executeDDLQueryOnCluster(query_ptr, context, {ast.database});
StoragePtr table = context.getTable(ast.database, ast.table);
auto table_lock = table->lockStructureForShare(true, context.getCurrentQueryId());
table->optimize(query_ptr, ast.partition, ast.final, ast.deduplicate, context);
return {};
}

206
dbms/src/Interpreters/InterpreterSelectQuery.cpp
View File

@ -82,6 +82,8 @@
#include <Processors/Transforms/RollupTransform.h>
#include <Processors/Transforms/CubeTransform.h>
#include <Processors/LimitTransform.h>
#include <DataTypes/DataTypeAggregateFunction.h>
#include <DataStreams/materializeBlock.h>
namespace DB
@ -271,7 +273,7 @@ InterpreterSelectQuery::InterpreterSelectQuery(
String database_name;
String table_name;
getDatabaseAndTableNames(database_name, table_name);
getDatabaseAndTableNames(query, database_name, table_name, context);
if (auto view_source = context.getViewSource())
{
@ -345,17 +347,20 @@ InterpreterSelectQuery::InterpreterSelectQuery(
source_header = storage->getSampleBlockForColumns(required_columns);
/// Calculate structure of the result.
result_header = getSampleBlockImpl();
for (auto & col : result_header)
{
Pipeline pipeline;
executeImpl(pipeline, nullptr, true);
result_header = pipeline.firstStream()->getHeader();
if (!col.column)
col.column = col.type->createColumn();
else if (isColumnConst(*col.column) && !col.column->empty())
col.column = col.column->cloneEmpty();
}
}
void InterpreterSelectQuery::getDatabaseAndTableNames(String & database_name, String & table_name)
void InterpreterSelectQuery::getDatabaseAndTableNames(const ASTSelectQuery & query, String & database_name, String & table_name, const Context & context)
{
if (auto db_and_table = getDatabaseAndTable(getSelectQuery(), 0))
if (auto db_and_table = getDatabaseAndTable(query, 0))
{
table_name = db_and_table->table;
database_name = db_and_table->database;
@ -381,8 +386,8 @@ Block InterpreterSelectQuery::getSampleBlock()
BlockIO InterpreterSelectQuery::execute()
{
Pipeline pipeline;
executeImpl(pipeline, input, options.only_analyze);
executeUnion(pipeline);
executeImpl(pipeline, input);
executeUnion(pipeline, getSampleBlock());
BlockIO res;
res.in = pipeline.firstStream();
@ -392,28 +397,104 @@ BlockIO InterpreterSelectQuery::execute()
BlockInputStreams InterpreterSelectQuery::executeWithMultipleStreams()
{
Pipeline pipeline;
executeImpl(pipeline, input, options.only_analyze);
executeImpl(pipeline, input);
unifyStreams(pipeline, getSampleBlock());
return pipeline.streams;
}
QueryPipeline InterpreterSelectQuery::executeWithProcessors()
{
QueryPipeline query_pipeline;
executeImpl(query_pipeline, input, options.only_analyze);
executeImpl(query_pipeline, input);
return query_pipeline;
}
Block InterpreterSelectQuery::getSampleBlockImpl()
{
FilterInfoPtr filter_info;
/// Need to create sets before analyzeExpressions(). Otherwise some sets for index won't be created.
query_analyzer->makeSetsForIndex(getSelectQuery().where());
query_analyzer->makeSetsForIndex(getSelectQuery().prewhere());
auto analysis_result = analyzeExpressions(
getSelectQuery(),
*query_analyzer,
QueryProcessingStage::Enum::FetchColumns,
options.to_stage,
context,
storage,
true,
filter_info);
if (options.to_stage == QueryProcessingStage::Enum::FetchColumns)
{
auto header = source_header;
if (analysis_result.prewhere_info)
{
analysis_result.prewhere_info->prewhere_actions->execute(header);
header = materializeBlock(header);
if (analysis_result.prewhere_info->remove_prewhere_column)
header.erase(analysis_result.prewhere_info->prewhere_column_name);
}
return header;
}
if (options.to_stage == QueryProcessingStage::Enum::WithMergeableState)
{
if (!analysis_result.need_aggregate)
return analysis_result.before_order_and_select->getSampleBlock();
auto header = analysis_result.before_aggregation->getSampleBlock();
Names key_names;
AggregateDescriptions aggregates;
query_analyzer->getAggregateInfo(key_names, aggregates);
Block res;
for (auto & key : key_names)
res.insert({nullptr, header.getByName(key).type, key});
for (auto & aggregate : aggregates)
{
size_t arguments_size = aggregate.argument_names.size();
DataTypes argument_types(arguments_size);
for (size_t j = 0; j < arguments_size; ++j)
argument_types[j] = header.getByName(aggregate.argument_names[j]).type;
DataTypePtr type = std::make_shared<DataTypeAggregateFunction>(aggregate.function, argument_types, aggregate.parameters);
res.insert({nullptr, type, aggregate.column_name});
}
return res;
}
return analysis_result.final_projection->getSampleBlock();
}
InterpreterSelectQuery::AnalysisResult
InterpreterSelectQuery::analyzeExpressions(QueryProcessingStage::Enum from_stage, bool dry_run, const FilterInfoPtr & filter_info)
InterpreterSelectQuery::analyzeExpressions(
const ASTSelectQuery & query,
SelectQueryExpressionAnalyzer & query_analyzer,
QueryProcessingStage::Enum from_stage,
QueryProcessingStage::Enum to_stage,
const Context & context,
const StoragePtr & storage,
bool only_types,
const FilterInfoPtr & filter_info)
{
AnalysisResult res;
/// Do I need to perform the first part of the pipeline - running on remote servers during distributed processing.
res.first_stage = from_stage < QueryProcessingStage::WithMergeableState
&& options.to_stage >= QueryProcessingStage::WithMergeableState;
&& to_stage >= QueryProcessingStage::WithMergeableState;
/// Do I need to execute the second part of the pipeline - running on the initiating server during distributed processing.
res.second_stage = from_stage <= QueryProcessingStage::WithMergeableState
&& options.to_stage > QueryProcessingStage::WithMergeableState;
&& to_stage > QueryProcessingStage::WithMergeableState;
/** First we compose a chain of actions and remember the necessary steps from it.
* Regardless of from_stage and to_stage, we will compose a complete sequence of actions to perform optimization and
@ -468,8 +549,6 @@ InterpreterSelectQuery::analyzeExpressions(QueryProcessingStage::Enum from_stage
{
ExpressionActionsChain chain(context);
auto & query = getSelectQuery();
Names additional_required_columns_after_prewhere;
if (storage && query.sample_size())
@ -486,14 +565,14 @@ InterpreterSelectQuery::analyzeExpressions(QueryProcessingStage::Enum from_stage
columns_for_final.begin(), columns_for_final.end());
}
if (storage && context.hasUserProperty(storage->getDatabaseName(), storage->getTableName(), "filter"))
if (storage && filter_info)
{
has_filter = true;
/// XXX: aggregated copy-paste from ExpressionAnalyzer::appendSmth()
if (chain.steps.empty())
{
chain.steps.emplace_back(std::make_shared<ExpressionActions>(source_columns, context));
chain.steps.emplace_back(std::make_shared<ExpressionActions>(NamesAndTypesList(), context));
}
ExpressionActionsChain::Step & step = chain.steps.back();
@ -506,7 +585,7 @@ InterpreterSelectQuery::analyzeExpressions(QueryProcessingStage::Enum from_stage
chain.addStep();
}
if (query_analyzer->appendPrewhere(chain, !res.first_stage, additional_required_columns_after_prewhere))
if (query_analyzer.appendPrewhere(chain, !res.first_stage, additional_required_columns_after_prewhere))
{
has_prewhere = true;
@ -516,11 +595,11 @@ InterpreterSelectQuery::analyzeExpressions(QueryProcessingStage::Enum from_stage
chain.addStep();
}
res.need_aggregate = query_analyzer->hasAggregation();
res.need_aggregate = query_analyzer.hasAggregation();
query_analyzer->appendArrayJoin(chain, dry_run || !res.first_stage);
query_analyzer.appendArrayJoin(chain, only_types || !res.first_stage);
if (query_analyzer->appendJoin(chain, dry_run || !res.first_stage))
if (query_analyzer.appendJoin(chain, only_types || !res.first_stage))
{
res.before_join = chain.getLastActions();
if (!res.hasJoin())
@ -528,7 +607,7 @@ InterpreterSelectQuery::analyzeExpressions(QueryProcessingStage::Enum from_stage
chain.addStep();
}
if (query_analyzer->appendWhere(chain, dry_run || !res.first_stage))
if (query_analyzer.appendWhere(chain, only_types || !res.first_stage))
{
where_step_num = chain.steps.size() - 1;
has_where = res.has_where = true;
@ -538,13 +617,13 @@ InterpreterSelectQuery::analyzeExpressions(QueryProcessingStage::Enum from_stage
if (res.need_aggregate)
{
query_analyzer->appendGroupBy(chain, dry_run || !res.first_stage);
query_analyzer->appendAggregateFunctionsArguments(chain, dry_run || !res.first_stage);
query_analyzer.appendGroupBy(chain, only_types || !res.first_stage);
query_analyzer.appendAggregateFunctionsArguments(chain, only_types || !res.first_stage);
res.before_aggregation = chain.getLastActions();
finalizeChain(chain);
if (query_analyzer->appendHaving(chain, dry_run || !res.second_stage))
if (query_analyzer.appendHaving(chain, only_types || !res.second_stage))
{
res.has_having = true;
res.before_having = chain.getLastActions();
@ -553,20 +632,20 @@ InterpreterSelectQuery::analyzeExpressions(QueryProcessingStage::Enum from_stage
}
/// If there is aggregation, we execute expressions in SELECT and ORDER BY on the initiating server, otherwise on the source servers.
query_analyzer->appendSelect(chain, dry_run || (res.need_aggregate ? !res.second_stage : !res.first_stage));
query_analyzer.appendSelect(chain, only_types || (res.need_aggregate ? !res.second_stage : !res.first_stage));
res.selected_columns = chain.getLastStep().required_output;
res.has_order_by = query_analyzer->appendOrderBy(chain, dry_run || (res.need_aggregate ? !res.second_stage : !res.first_stage));
res.has_order_by = query_analyzer.appendOrderBy(chain, only_types || (res.need_aggregate ? !res.second_stage : !res.first_stage));
res.before_order_and_select = chain.getLastActions();
chain.addStep();
if (query_analyzer->appendLimitBy(chain, dry_run || !res.second_stage))
if (query_analyzer.appendLimitBy(chain, only_types || !res.second_stage))
{
res.has_limit_by = true;
res.before_limit_by = chain.getLastActions();
chain.addStep();
}
query_analyzer->appendProjectResult(chain);
query_analyzer.appendProjectResult(chain);
res.final_projection = chain.getLastActions();
finalizeChain(chain);
@ -580,7 +659,7 @@ InterpreterSelectQuery::analyzeExpressions(QueryProcessingStage::Enum from_stage
if (res.has_having)
res.before_having->prependProjectInput();
res.subqueries_for_sets = query_analyzer->getSubqueriesForSets();
res.subqueries_for_sets = query_analyzer.getSubqueriesForSets();
/// Check that PREWHERE doesn't contain unusual actions. Unusual actions are that can change number of rows.
if (res.prewhere_info)
@ -747,7 +826,7 @@ static SortingInfoPtr optimizeReadInOrder(const MergeTreeData & merge_tree, cons
template <typename TPipeline>
void InterpreterSelectQuery::executeImpl(TPipeline & pipeline, const BlockInputStreamPtr & prepared_input, bool dry_run)
void InterpreterSelectQuery::executeImpl(TPipeline & pipeline, const BlockInputStreamPtr & prepared_input)
{
/** Streams of data. When the query is executed in parallel, we have several data streams.
* If there is no GROUP BY, then perform all operations before ORDER BY and LIMIT in parallel, then
@ -771,7 +850,7 @@ void InterpreterSelectQuery::executeImpl(TPipeline & pipeline, const BlockInputS
/// Turn off, if the table filter is applied.
if (storage && !context.hasUserProperty(storage->getDatabaseName(), storage->getTableName(), "filter"))
{
if (!dry_run)
if (!options.only_analyze)
from_stage = storage->getQueryProcessingStage(context);
query_analyzer->makeSetsForIndex(query.where());
@ -811,14 +890,22 @@ void InterpreterSelectQuery::executeImpl(TPipeline & pipeline, const BlockInputS
sorting_info = optimizeReadInOrder(*merge_tree_data, query, context, syntax_analyzer_result);
}
if (dry_run)
if (options.only_analyze)
{
if constexpr (pipeline_with_processors)
pipeline.init({std::make_shared<NullSource>(source_header)});
else
pipeline.streams.emplace_back(std::make_shared<NullBlockInputStream>(source_header));
expressions = analyzeExpressions(QueryProcessingStage::FetchColumns, true, filter_info);
expressions = analyzeExpressions(
getSelectQuery(),
*query_analyzer,
QueryProcessingStage::FetchColumns,
options.to_stage,
context,
storage,
true,
filter_info);
if (storage && expressions.filter_info && expressions.prewhere_info)
throw Exception("PREWHERE is not supported if the table is filtered by row-level security expression", ErrorCodes::ILLEGAL_PREWHERE);
@ -850,7 +937,15 @@ void InterpreterSelectQuery::executeImpl(TPipeline & pipeline, const BlockInputS
pipeline.streams.push_back(prepared_input);
}
expressions = analyzeExpressions(from_stage, false, filter_info);
expressions = analyzeExpressions(
getSelectQuery(),
*query_analyzer,
from_stage,
options.to_stage,
context,
storage,
false,
filter_info);
if (from_stage == QueryProcessingStage::WithMergeableState &&
options.to_stage == QueryProcessingStage::WithMergeableState)
@ -1097,7 +1192,7 @@ void InterpreterSelectQuery::executeImpl(TPipeline & pipeline, const BlockInputS
if constexpr (pipeline_with_processors)
pipeline.resize(1);
else
executeUnion(pipeline);
executeUnion(pipeline, {});
}
/** If there was more than one stream,
@ -1696,7 +1791,7 @@ void InterpreterSelectQuery::executeMergeAggregated(Pipeline & pipeline, bool ov
if (!settings.distributed_aggregation_memory_efficient)
{
/// We union several sources into one, parallelizing the work.
executeUnion(pipeline);
executeUnion(pipeline, {});
/// Now merge the aggregated blocks
pipeline.firstStream() = std::make_shared<MergingAggregatedBlockInputStream>(pipeline.firstStream(), params, final, settings.max_threads);
@ -1798,7 +1893,7 @@ void InterpreterSelectQuery::executeHaving(QueryPipeline & pipeline, const Expre
void InterpreterSelectQuery::executeTotalsAndHaving(Pipeline & pipeline, bool has_having, const ExpressionActionsPtr & expression, bool overflow_row, bool final)
{
executeUnion(pipeline);
executeUnion(pipeline, {});
const Settings & settings = context.getSettingsRef();
@ -1827,7 +1922,7 @@ void InterpreterSelectQuery::executeTotalsAndHaving(QueryPipeline & pipeline, bo
void InterpreterSelectQuery::executeRollupOrCube(Pipeline & pipeline, Modificator modificator)
{
executeUnion(pipeline);
executeUnion(pipeline, {});
Names key_names;
AggregateDescriptions aggregates;
@ -1972,7 +2067,7 @@ void InterpreterSelectQuery::executeOrder(Pipeline & pipeline, SortingInfoPtr so
});
/// If there are several streams, we merge them into one
executeUnion(pipeline);
executeUnion(pipeline, {});
/// Merge the sorted blocks.
pipeline.firstStream() = std::make_shared<MergeSortingBlockInputStream>(
@ -2032,7 +2127,7 @@ void InterpreterSelectQuery::executeMergeSorted(Pipeline & pipeline)
/// If there are several streams, then we merge them into one
if (pipeline.hasMoreThanOneStream())
{
unifyStreams(pipeline);
unifyStreams(pipeline, pipeline.firstStream()->getHeader());
/** MergingSortedBlockInputStream reads the sources sequentially.
* To make the data on the remote servers prepared in parallel, we wrap it in AsynchronousBlockInputStream.
@ -2136,12 +2231,15 @@ void InterpreterSelectQuery::executeDistinct(QueryPipeline & pipeline, bool befo
}
void InterpreterSelectQuery::executeUnion(Pipeline & pipeline)
void InterpreterSelectQuery::executeUnion(Pipeline & pipeline, Block header)
{
/// If there are still several streams, then we combine them into one
if (pipeline.hasMoreThanOneStream())
{
unifyStreams(pipeline);
if (!header)
header = pipeline.firstStream()->getHeader();
unifyStreams(pipeline, std::move(header));
pipeline.firstStream() = std::make_shared<UnionBlockInputStream>(pipeline.streams, pipeline.stream_with_non_joined_data, max_streams);
pipeline.stream_with_non_joined_data = nullptr;
@ -2351,7 +2449,7 @@ void InterpreterSelectQuery::executeExtremes(QueryPipeline & pipeline)
void InterpreterSelectQuery::executeSubqueriesInSetsAndJoins(Pipeline & pipeline, SubqueriesForSets & subqueries_for_sets)
{
executeUnion(pipeline);
executeUnion(pipeline, {});
pipeline.firstStream() = std::make_shared<CreatingSetsBlockInputStream>(
pipeline.firstStream(), subqueries_for_sets, context);
}
@ -2369,20 +2467,22 @@ void InterpreterSelectQuery::executeSubqueriesInSetsAndJoins(QueryPipeline & pip
}
void InterpreterSelectQuery::unifyStreams(Pipeline & pipeline)
void InterpreterSelectQuery::unifyStreams(Pipeline & pipeline, Block header)
{
if (pipeline.hasMoreThanOneStream())
{
/// Unify streams in case they have different headers.
auto first_header = pipeline.streams.at(0)->getHeader();
for (size_t i = 1; i < pipeline.streams.size(); ++i)
/// TODO: remove previos addition of _dummy column.
if (header.columns() > 1 && header.has("_dummy"))
header.erase("_dummy");
for (size_t i = 0; i < pipeline.streams.size(); ++i)
{
auto & stream = pipeline.streams[i];
auto header = stream->getHeader();
auto stream_header = stream->getHeader();
auto mode = ConvertingBlockInputStream::MatchColumnsMode::Name;
if (!blocksHaveEqualStructure(first_header, header))
stream = std::make_shared<ConvertingBlockInputStream>(context, stream, first_header, mode);
}
if (!blocksHaveEqualStructure(header, stream_header))
stream = std::make_shared<ConvertingBlockInputStream>(context, stream, header, mode);
}
}

23
dbms/src/Interpreters/InterpreterSelectQuery.h
View File

@ -90,6 +90,7 @@ private:
ASTSelectQuery & getSelectQuery() { return query_ptr->as<ASTSelectQuery &>(); }
Block getSampleBlockImpl();
struct Pipeline
{
@ -135,7 +136,7 @@ private:
};
template <typename TPipeline>
void executeImpl(TPipeline & pipeline, const BlockInputStreamPtr & prepared_input, bool dry_run);
void executeImpl(TPipeline & pipeline, const BlockInputStreamPtr & prepared_input);
struct AnalysisResult
{
@ -172,12 +173,19 @@ private:
FilterInfoPtr filter_info;
};
AnalysisResult analyzeExpressions(QueryProcessingStage::Enum from_stage, bool dry_run, const FilterInfoPtr & filter_info);
static AnalysisResult analyzeExpressions(
const ASTSelectQuery & query,
SelectQueryExpressionAnalyzer & query_analyzer,
QueryProcessingStage::Enum from_stage,
QueryProcessingStage::Enum to_stage,
const Context & context,
const StoragePtr & storage,
bool only_types,
const FilterInfoPtr & filter_info);
/** From which table to read. With JOIN, the "left" table is returned.
*/
void getDatabaseAndTableNames(String & database_name, String & table_name);
static void getDatabaseAndTableNames(const ASTSelectQuery & query, String & database_name, String & table_name, const Context & context);
/// Different stages of query execution.
@ -198,7 +206,7 @@ private:
void executeOrder(Pipeline & pipeline, SortingInfoPtr sorting_info);
void executeMergeSorted(Pipeline & pipeline);
void executePreLimit(Pipeline & pipeline);
void executeUnion(Pipeline & pipeline);
void executeUnion(Pipeline & pipeline, Block header); /// If header is not empty, convert streams structure to it.
void executeLimitBy(Pipeline & pipeline);
void executeLimit(Pipeline & pipeline);
void executeProjection(Pipeline & pipeline, const ExpressionActionsPtr & expression);
@ -222,8 +230,8 @@ private:
void executeExtremes(QueryPipeline & pipeline);
void executeSubqueriesInSetsAndJoins(QueryPipeline & pipeline, std::unordered_map<String, SubqueryForSet> & subqueries_for_sets);
/// If pipeline has several streams with different headers, add ConvertingBlockInputStream to first header.
void unifyStreams(Pipeline & pipeline);
/// Add ConvertingBlockInputStream to specified header.
void unifyStreams(Pipeline & pipeline, Block header);
enum class Modificator
{
@ -246,7 +254,6 @@ private:
const SelectQueryOptions options;
ASTPtr query_ptr;
Context context;
NamesAndTypesList source_columns;
SyntaxAnalyzerResultPtr syntax_analyzer_result;
std::unique_ptr<SelectQueryExpressionAnalyzer> query_analyzer;
SelectQueryInfo query_info;

108
dbms/src/Interpreters/InterpreterWatchQuery.cpp Normal file
View File

@ -0,0 +1,108 @@
/* Copyright (c) 2018 BlackBerry Limited
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License. */
#include <Core/Settings.h>
#include <Common/typeid_cast.h>
#include <Parsers/ASTWatchQuery.h>
#include <Interpreters/InterpreterWatchQuery.h>
#include <DataStreams/IBlockInputStream.h>
#include <DataStreams/OneBlockInputStream.h>
namespace DB
{
namespace ErrorCodes
{
extern const int UNKNOWN_STORAGE;
extern const int UNKNOWN_TABLE;
extern const int TOO_MANY_COLUMNS;
}
BlockInputStreamPtr InterpreterWatchQuery::executeImpl()
{
return std::make_shared<OneBlockInputStream>(Block());
}
BlockIO InterpreterWatchQuery::execute()
{
BlockIO res;
const ASTWatchQuery & query = typeid_cast<const ASTWatchQuery &>(*query_ptr);
String database;
String table;
/// Get database
if (!query.database.empty())
database = query.database;
else
database = context.getCurrentDatabase();
/// Get table
table = query.table;
/// Get storage
storage = context.tryGetTable(database, table);
if (!storage)
throw Exception("Table " + backQuoteIfNeed(database) + "." +
backQuoteIfNeed(table) + " doesn't exist.",
ErrorCodes::UNKNOWN_TABLE);
/// List of columns to read to execute the query.
Names required_columns = storage->getColumns().getNamesOfPhysical();
/// Get context settings for this query
const Settings & settings = context.getSettingsRef();
/// Limitation on the number of columns to read.
if (settings.max_columns_to_read && required_columns.size() > settings.max_columns_to_read)
throw Exception("Limit for number of columns to read exceeded. "
"Requested: " + std::to_string(required_columns.size())
+ ", maximum: " + settings.max_columns_to_read.toString(),
ErrorCodes::TOO_MANY_COLUMNS);
size_t max_block_size = settings.max_block_size;
size_t max_streams = 1;
/// Define query info
SelectQueryInfo query_info;
query_info.query = query_ptr;
/// From stage
QueryProcessingStage::Enum from_stage = QueryProcessingStage::FetchColumns;
QueryProcessingStage::Enum to_stage = QueryProcessingStage::Complete;
/// Watch storage
streams = storage->watch(required_columns, query_info, context, from_stage, max_block_size, max_streams);
/// Constraints on the result, the quota on the result, and also callback for progress.
if (IBlockInputStream * stream = dynamic_cast<IBlockInputStream *>(streams[0].get()))
{
/// Constraints apply only to the final result.
if (to_stage == QueryProcessingStage::Complete)
{
IBlockInputStream::LocalLimits limits;
limits.mode = IBlockInputStream::LIMITS_CURRENT;
limits.size_limits.max_rows = settings.max_result_rows;
limits.size_limits.max_bytes = settings.max_result_bytes;
limits.size_limits.overflow_mode = settings.result_overflow_mode;
stream->setLimits(limits);
stream->setQuota(context.getQuota());
}
}
res.in = streams[0];
return res;
}
}

50
dbms/src/Interpreters/InterpreterWatchQuery.h Normal file
View File

@ -0,0 +1,50 @@
/* Copyright (c) 2018 BlackBerry Limited
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License. */
#pragma once
#include <Core/QueryProcessingStage.h>
#include <DataStreams/BlockIO.h>
#include <DataStreams/IBlockInputStream.h>
#include <Parsers/IAST_fwd.h>
#include <Interpreters/IInterpreter.h>
#include <Storages/SelectQueryInfo.h>
#include <Storages/IStorage.h>
#include <Interpreters/Context.h>
namespace DB
{
class IAST;
using ASTPtr = std::shared_ptr<IAST>;
using StoragePtr = std::shared_ptr<IStorage>;
class InterpreterWatchQuery : public IInterpreter
{
public:
InterpreterWatchQuery(const ASTPtr & query_ptr_, Context & context_)
: query_ptr(query_ptr_), context(context_) {}
BlockIO execute() override;
private:
ASTPtr query_ptr;
Context & context;
BlockInputStreamPtr executeImpl();
/// Table from where to read data, if not subquery.
StoragePtr storage;
/// Streams of read data
BlockInputStreams streams;
};
}

3
dbms/src/Interpreters/Join.h
View File

@ -126,6 +126,8 @@ public:
bool empty() { return type == Type::EMPTY; }
bool isNullUsedAsDefault() const { return use_nulls; }
/** Set information about structure of right hand of JOIN (joined data).
* You must call this method before subsequent calls to insertFromBlock.
*/
@ -168,6 +170,7 @@ public:
size_t getTotalByteCount() const;
ASTTableJoin::Kind getKind() const { return kind; }
ASTTableJoin::Strictness getStrictness() const { return strictness; }
AsofRowRefs::Type getAsofType() const { return *asof_type; }
bool anyTakeLastRow() const { return any_take_last_row; }

7
dbms/src/Interpreters/MutationsInterpreter.cpp
View File

@ -458,15 +458,16 @@ BlockInputStreamPtr MutationsInterpreter::addStreamsForLaterStages(const std::ve
return in;
}
void MutationsInterpreter::validate()
void MutationsInterpreter::validate(TableStructureReadLockHolder &)
{
prepare(/* dry_run = */ true);
Block first_stage_header = interpreter_select->getSampleBlock();
/// Do not use getSampleBlock in order to check the whole pipeline.
Block first_stage_header = interpreter_select->execute().in->getHeader();
BlockInputStreamPtr in = std::make_shared<NullBlockInputStream>(first_stage_header);
addStreamsForLaterStages(stages, in)->getHeader();
}
BlockInputStreamPtr MutationsInterpreter::execute()
BlockInputStreamPtr MutationsInterpreter::execute(TableStructureReadLockHolder &)
{
prepare(/* dry_run = */ false);
BlockInputStreamPtr in = interpreter_select->execute().in;

5
dbms/src/Interpreters/MutationsInterpreter.h
View File

@ -25,13 +25,13 @@ public:
{
}
void validate();
void validate(TableStructureReadLockHolder & table_lock_holder);
/// Return false if the data isn't going to be changed by mutations.
bool isStorageTouchedByMutations() const;
/// The resulting stream will return blocks containing only changed columns and columns, that we need to recalculate indices.
BlockInputStreamPtr execute();
BlockInputStreamPtr execute(TableStructureReadLockHolder & table_lock_holder);
/// Only changed columns.
const Block & getUpdatedHeader() const;
@ -44,7 +44,6 @@ private:
std::unique_ptr<InterpreterSelectQuery> prepareInterpreterSelect(std::vector<Stage> & prepared_stages, bool dry_run);
BlockInputStreamPtr addStreamsForLaterStages(const std::vector<Stage> & prepared_stages, BlockInputStreamPtr in) const;
private:
StoragePtr storage;
std::vector<MutationCommand> commands;
const Context & context;

18
dbms/src/Interpreters/QueryNormalizer.cpp
View File

@ -102,8 +102,24 @@ void QueryNormalizer::visit(ASTIdentifier & node, ASTPtr & ast, Data & data)
/// If it is an alias, but not a parent alias (for constructs like "SELECT column + 1 AS column").
auto it_alias = data.aliases.find(node.name);
if (IdentifierSemantic::canBeAlias(node) && it_alias != data.aliases.end() && current_alias != node.name)
if (it_alias != data.aliases.end() && current_alias != node.name)
{
if (!IdentifierSemantic::canBeAlias(node))
{
/// This means that column had qualified name, which was translated (so, canBeAlias() returns false).
/// But there is an alias with the same name. So, let's use original name for that column.
/// If alias wasn't set, use original column name as alias.
/// That helps to avoid result set with columns which have same names but different values.
if (node.alias.empty())
{
node.name.swap(node.alias);
node.restoreCompoundName();
node.name.swap(node.alias);
}
return;
}
auto & alias_node = it_alias->second;
/// Let's replace it with the corresponding tree node.

32
dbms/src/Interpreters/Set.cpp
View File

@ -320,13 +320,7 @@ ColumnPtr Set::execute(const Block & block, bool negative) const
return res;
}
if (data_types.size() != num_key_columns)
{
std::stringstream message;
message << "Number of columns in section IN doesn't match. "
<< num_key_columns << " at left, " << data_types.size() << " at right.";
throw Exception(message.str(), ErrorCodes::NUMBER_OF_COLUMNS_DOESNT_MATCH);
}
checkColumnsNumber(num_key_columns);
/// Remember the columns we will work with. Also check that the data types are correct.
ColumnRawPtrs key_columns;
@ -337,11 +331,7 @@ ColumnPtr Set::execute(const Block & block, bool negative) const
for (size_t i = 0; i < num_key_columns; ++i)
{
if (!removeNullable(data_types[i])->equals(*removeNullable(block.safeGetByPosition(i).type)))
throw Exception("Types of column " + toString(i + 1) + " in section IN don't match: "
+ data_types[i]->getName() + " on the right, " + block.safeGetByPosition(i).type->getName() +
" on the left.", ErrorCodes::TYPE_MISMATCH);
checkTypesEqual(i, block.safeGetByPosition(i).type);
materialized_columns.emplace_back(block.safeGetByPosition(i).column->convertToFullColumnIfConst());
key_columns.emplace_back() = materialized_columns.back().get();
}
@ -421,6 +411,24 @@ void Set::executeOrdinary(
}
}
void Set::checkColumnsNumber(size_t num_key_columns) const
{
if (data_types.size() != num_key_columns)
{
std::stringstream message;
message << "Number of columns in section IN doesn't match. "
<< num_key_columns << " at left, " << data_types.size() << " at right.";
throw Exception(message.str(), ErrorCodes::NUMBER_OF_COLUMNS_DOESNT_MATCH);
}
}
void Set::checkTypesEqual(size_t set_type_idx, const DataTypePtr & other_type) const
{
if (!removeNullable(data_types[set_type_idx])->equals(*removeNullable(other_type)))
throw Exception("Types of column " + toString(set_type_idx + 1) + " in section IN don't match: "
+ data_types[set_type_idx]->getName() + " on the right, " + other_type->getName() +
" on the left.", ErrorCodes::TYPE_MISMATCH);
}
MergeTreeSetIndex::MergeTreeSetIndex(const Columns & set_elements, std::vector<KeyTuplePositionMapping> && index_mapping_)
: indexes_mapping(std::move(index_mapping_))

3
dbms/src/Interpreters/Set.h
View File

@ -70,6 +70,9 @@ public:
bool hasExplicitSetElements() const { return fill_set_elements; }
Columns getSetElements() const { return { set_elements.begin(), set_elements.end() }; }
void checkColumnsNumber(size_t num_key_columns) const;
void checkTypesEqual(size_t set_type_idx, const DataTypePtr & other_type) const;
private:
size_t keys_size = 0;
Sizes key_sizes;

41
dbms/src/NOTICE Normal file
View File

@ -0,0 +1,41 @@
--
The following notice shall be applied to the files listed below.
Some modifications Copyright (c) 2018 BlackBerry Limited
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
Common/ErrorCodes.cpp
Common/UInt128.h
Core/Block.h
Core/Defines.h
Core/Settings.h
DataStreams/PushingToViewsBlockOutputStream.cpp
DataStreams/PushingToViewsBlockOutputStream.h
DataStreams/copyData.cpp
Databases/DatabasesCommon.cpp
IO/WriteBufferValidUTF8.cpp
Interpreters/InterpreterAlterQuery.cpp
Interpreters/InterpreterCreateQuery.cpp
Interpreters/InterpreterFactory.cpp
Parsers/ASTAlterQuery.cpp
Parsers/ASTAlterQuery.h
Parsers/ASTCreateQuery.cpp
Parsers/ASTCreateQuery.h
Parsers/ParserAlterQuery.cpp
Parsers/ParserAlterQuery.h
Parsers/ParserCreateQuery.cpp
Parsers/ParserCreateQuery.h
Parsers/ParserQueryWithOutput.cpp
Storages/IStorage.h
Storages/StorageFactory.cpp
Storages/registerStorages.cpp
--

50
dbms/src/Parsers/ASTAlterQuery.cpp
View File

@ -50,6 +50,11 @@ ASTPtr ASTAlterCommand::clone() const
res->settings_changes = settings_changes->clone();
res->children.push_back(res->settings_changes);
}
if (values)
{
res->values = values->clone();
res->children.push_back(res->values);
}
return res;
}
@ -210,6 +215,46 @@ void ASTAlterCommand::formatImpl(
settings.ostr << (settings.hilite ? hilite_keyword : "") << indent_str << "MODIFY SETTING " << (settings.hilite ? hilite_none : "");
settings_changes->formatImpl(settings, state, frame);
}
else if (type == ASTAlterCommand::LIVE_VIEW_REFRESH)
{
settings.ostr << (settings.hilite ? hilite_keyword : "") << indent_str << "REFRESH " << (settings.hilite ? hilite_none : "");
}
else if (type == ASTAlterCommand::LIVE_CHANNEL_ADD)
{
settings.ostr << (settings.hilite ? hilite_keyword : "") << indent_str << "ADD " << (settings.hilite ? hilite_none : "");
values->formatImpl(settings, state, frame);
}
else if (type == ASTAlterCommand::LIVE_CHANNEL_DROP)
{
settings.ostr << (settings.hilite ? hilite_keyword : "") << indent_str << "DROP " << (settings.hilite ? hilite_none : "");
values->formatImpl(settings, state, frame);
}
else if (type == ASTAlterCommand::LIVE_CHANNEL_MODIFY)
{
settings.ostr << (settings.hilite ? hilite_keyword : "") << indent_str << "MODIFY " << (settings.hilite ? hilite_none : "");
values->formatImpl(settings, state, frame);
}
else if (type == ASTAlterCommand::LIVE_CHANNEL_SUSPEND)
{
settings.ostr << (settings.hilite ? hilite_keyword : "") << indent_str << "SUSPEND " << (settings.hilite ? hilite_none : "");
values->formatImpl(settings, state, frame);
}
else if (type == ASTAlterCommand::LIVE_CHANNEL_RESUME)
{
settings.ostr << (settings.hilite ? hilite_keyword : "") << indent_str << "RESUME " << (settings.hilite ? hilite_none : "");
values->formatImpl(settings, state, frame);
}
else if (type == ASTAlterCommand::LIVE_CHANNEL_REFRESH)
{
settings.ostr << (settings.hilite ? hilite_keyword : "") << indent_str << "REFRESH " << (settings.hilite ? hilite_none : "");
values->formatImpl(settings, state, frame);
}
else
throw Exception("Unexpected type of ALTER", ErrorCodes::UNEXPECTED_AST_STRUCTURE);
}
@ -262,6 +307,11 @@ void ASTAlterQuery::formatQueryImpl(const FormatSettings & settings, FormatState
std::string indent_str = settings.one_line ? "" : std::string(4u * frame.indent, ' ');
if (is_live_view)
settings.ostr << (settings.hilite ? hilite_keyword : "") << indent_str << "ALTER LIVE VIEW " << (settings.hilite ? hilite_none : "");
else if (is_live_channel)
settings.ostr << (settings.hilite ? hilite_keyword : "") << indent_str << "ALTER LIVE CHANNEL " << (settings.hilite ? hilite_none : "");
else
settings.ostr << (settings.hilite ? hilite_keyword : "") << indent_str << "ALTER TABLE " << (settings.hilite ? hilite_none : "");
if (!table.empty())

25
dbms/src/Parsers/ASTAlterQuery.h
View File

@ -15,6 +15,15 @@ namespace DB
* MODIFY COLUMN col_name type,
* DROP PARTITION partition,
* COMMENT_COLUMN col_name 'comment',
* ALTER LIVE VIEW [db.]name_type
* REFRESH
* ALTER CHANNEL [db.]name_type
* ADD live_view,...
* DROP live_view,...
* SUSPEND live_view,...
* RESUME live_view,...
* REFRESH live_view,...
* MODIFY live_view,...
*/
class ASTAlterCommand : public IAST
@ -45,6 +54,15 @@ public:
UPDATE,
NO_TYPE,
LIVE_VIEW_REFRESH,
LIVE_CHANNEL_ADD,
LIVE_CHANNEL_DROP,
LIVE_CHANNEL_SUSPEND,
LIVE_CHANNEL_RESUME,
LIVE_CHANNEL_REFRESH,
LIVE_CHANNEL_MODIFY
};
Type type = NO_TYPE;
@ -95,6 +113,10 @@ public:
/// FOR MODIFY_SETTING
ASTPtr settings_changes;
/** In ALTER CHANNEL, ADD, DROP, SUSPEND, RESUME, REFRESH, MODIFY queries, the list of live views is stored here
*/
ASTPtr values;
bool detach = false; /// true for DETACH PARTITION
bool part = false; /// true for ATTACH PART
@ -151,6 +173,9 @@ protected:
class ASTAlterQuery : public ASTQueryWithTableAndOutput, public ASTQueryWithOnCluster
{
public:
bool is_live_view{false}; /// true for ALTER LIVE VIEW
bool is_live_channel{false}; /// true for ALTER LIVE CHANNEL
ASTAlterCommandList * command_list = nullptr;
String getID(char) const override;

13
dbms/src/Parsers/ASTCreateQuery.cpp
View File

@ -173,6 +173,8 @@ ASTPtr ASTCreateQuery::clone() const
res->set(res->storage, storage->clone());
if (select)
res->set(res->select, select->clone());
if (tables)
res->set(res->tables, tables->clone());
cloneOutputOptions(*res);
@ -204,6 +206,11 @@ void ASTCreateQuery::formatQueryImpl(const FormatSettings & settings, FormatStat
what = "VIEW";
if (is_materialized_view)
what = "MATERIALIZED VIEW";
if (is_live_view)
what = "LIVE VIEW";
if (is_live_channel)
what = "LIVE CHANNEL";
settings.ostr
<< (settings.hilite ? hilite_keyword : "")
@ -257,6 +264,12 @@ void ASTCreateQuery::formatQueryImpl(const FormatSettings & settings, FormatStat
settings.ostr << (settings.hilite ? hilite_keyword : "") << " AS" << settings.nl_or_ws << (settings.hilite ? hilite_none : "");
select->formatImpl(settings, state, frame);
}
if (tables)
{
settings.ostr << (settings.hilite ? hilite_keyword : "") << " WITH " << (settings.hilite ? hilite_none : "");
tables->formatImpl(settings, state, frame);
}
}
}

3
dbms/src/Parsers/ASTCreateQuery.h
View File

@ -55,9 +55,12 @@ public:
bool if_not_exists{false};
bool is_view{false};
bool is_materialized_view{false};
bool is_live_view{false};
bool is_live_channel{false};
bool is_populate{false};
bool replace_view{false}; /// CREATE OR REPLACE VIEW
ASTColumns * columns_list = nullptr;
ASTExpressionList *tables = nullptr;
String to_database; /// For CREATE MATERIALIZED VIEW mv TO table.
String to_table;
ASTStorage * storage = nullptr;

59
dbms/src/Parsers/ASTWatchQuery.h Normal file
View File

@ -0,0 +1,59 @@
/* Copyright (c) 2018 BlackBerry Limited
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License. */
#pragma once
#include <Parsers/ASTQueryWithTableAndOutput.h>
namespace DB
{
class ASTWatchQuery : public ASTQueryWithTableAndOutput
{
public:
ASTPtr limit_length;
bool is_watch_events;
ASTWatchQuery() = default;
String getID(char) const override { return "WatchQuery_" + database + "_" + table; }
ASTPtr clone() const override
{
std::shared_ptr<ASTWatchQuery> res = std::make_shared<ASTWatchQuery>(*this);
res->children.clear();
cloneOutputOptions(*res);
return res;
}
protected:
void formatQueryImpl(const FormatSettings & s, FormatState & state, FormatStateStacked frame) const override
{
std::string indent_str = s.one_line ? "" : std::string(4 * frame.indent, ' ');
s.ostr << (s.hilite ? hilite_keyword : "") << "WATCH" << " " << (s.hilite ? hilite_none : "")
<< (!database.empty() ? backQuoteIfNeed(database) + "." : "") << backQuoteIfNeed(table);
if (is_watch_events)
{
s.ostr << " " << (s.hilite ? hilite_keyword : "") << "EVENTS" << (s.hilite ? hilite_none : "");
}
if (limit_length)
{
s.ostr << (s.hilite ? hilite_keyword : "") << s.nl_or_ws << indent_str << "LIMIT " << (s.hilite ? hilite_none : "");
limit_length->formatImpl(s, state, frame);
}
}
};
}

98
dbms/src/Parsers/ParserAlterQuery.cpp
View File

@ -36,6 +36,13 @@ bool ParserAlterCommand::parseImpl(Pos & pos, ASTPtr & node, Expected & expected
ParserKeyword s_clear_index("CLEAR INDEX");
ParserKeyword s_materialize_index("MATERIALIZE INDEX");
ParserKeyword s_add("ADD");
ParserKeyword s_drop("DROP");
ParserKeyword s_suspend("SUSPEND");
ParserKeyword s_resume("RESUME");
ParserKeyword s_refresh("REFRESH");
ParserKeyword s_modify("MODIFY");
ParserKeyword s_attach_partition("ATTACH PARTITION");
ParserKeyword s_detach_partition("DETACH PARTITION");
ParserKeyword s_drop_partition("DROP PARTITION");
@ -68,7 +75,66 @@ bool ParserAlterCommand::parseImpl(Pos & pos, ASTPtr & node, Expected & expected
std::make_unique<ParserAssignment>(), std::make_unique<ParserToken>(TokenType::Comma),
/* allow_empty = */ false);
ParserSetQuery parser_settings(true);
ParserNameList values_p;
if (is_live_view)
{
if (s_refresh.ignore(pos, expected))
{
command->type = ASTAlterCommand::LIVE_VIEW_REFRESH;
}
else
return false;
}
else if (is_live_channel)
{
if (s_add.ignore(pos, expected))
{
if (!values_p.parse(pos, command->values, expected))
return false;
command->type = ASTAlterCommand::LIVE_CHANNEL_ADD;
}
else if (s_drop.ignore(pos, expected))
{
if (!values_p.parse(pos, command->values, expected))
return false;
command->type = ASTAlterCommand::LIVE_CHANNEL_DROP;
}
else if (s_suspend.ignore(pos, expected))
{
if (!values_p.parse(pos, command->values, expected))
return false;
command->type = ASTAlterCommand::LIVE_CHANNEL_SUSPEND;
}
else if (s_resume.ignore(pos, expected))
{
if (!values_p.parse(pos, command->values, expected))
return false;
command->type = ASTAlterCommand::LIVE_CHANNEL_RESUME;
}
else if (s_refresh.ignore(pos, expected))
{
if (!values_p.parse(pos, command->values, expected))
return false;
command->type = ASTAlterCommand::LIVE_CHANNEL_REFRESH;
}
else if (s_modify.ignore(pos, expected))
{
if (!values_p.parse(pos, command->values, expected))
return false;
command->type = ASTAlterCommand::LIVE_CHANNEL_MODIFY;
}
else
return false;
}
else
{
if (s_add_column.ignore(pos, expected))
{
if (s_if_not_exists.ignore(pos, expected))
@ -337,6 +403,8 @@ bool ParserAlterCommand::parseImpl(Pos & pos, ASTPtr & node, Expected & expected
else
return false;
}
if (command->col_decl)
command->children.push_back(command->col_decl);
if (command->column)
@ -349,10 +417,14 @@ bool ParserAlterCommand::parseImpl(Pos & pos, ASTPtr & node, Expected & expected
command->children.push_back(command->predicate);
if (command->update_assignments)
command->children.push_back(command->update_assignments);
if (command->values)
command->children.push_back(command->values);
if (command->comment)
command->children.push_back(command->comment);
if (command->ttl)
command->children.push_back(command->ttl);
if (command->settings_changes)
command->children.push_back(command->settings_changes);
return true;
}
@ -364,7 +436,7 @@ bool ParserAlterCommandList::parseImpl(Pos & pos, ASTPtr & node, Expected & expe
node = command_list;
ParserToken s_comma(TokenType::Comma);
ParserAlterCommand p_command;
ParserAlterCommand p_command(is_live_view, is_live_channel);
do
{
@ -413,8 +485,30 @@ bool ParserAlterQuery::parseImpl(Pos & pos, ASTPtr & node, Expected & expected)
node = query;
ParserKeyword s_alter_table("ALTER TABLE");
ParserKeyword s_alter_live_view("ALTER LIVE VIEW");
ParserKeyword s_alter_live_channel("ALTER LIVE CHANNEL");
bool is_live_view = false;
bool is_live_channel = false;
if (!s_alter_table.ignore(pos, expected))
{
if (!s_alter_live_view.ignore(pos, expected))
{
if (!s_alter_live_channel.ignore(pos, expected))
return false;
else
is_live_channel = true;
}
else
is_live_view = true;
}
if (is_live_view)
query->is_live_view = true;
if (is_live_channel)
query->is_live_channel = true;
if (!parseDatabaseAndTableName(pos, expected, query->database, query->table))
return false;
@ -427,7 +521,7 @@ bool ParserAlterQuery::parseImpl(Pos & pos, ASTPtr & node, Expected & expected)
}
query->cluster = cluster_str;
ParserAlterCommandList p_command_list;
ParserAlterCommandList p_command_list(is_live_view, is_live_channel);
ASTPtr command_list;
if (!p_command_list.parse(pos, command_list, expected))
return false;

21
dbms/src/Parsers/ParserAlterQuery.h
View File

@ -20,6 +20,15 @@ namespace DB
* [FREEZE [PARTITION] [WITH NAME name]]
* [DELETE WHERE ...]
* [UPDATE col_name = expr, ... WHERE ...]
* ALTER LIVE VIEW [db.name]
* [REFRESH]
* ALTER LIVE CHANNEL [db.name] [ON CLUSTER cluster]
* [ADD live_view, ...]
* [DROP live_view, ...]
* [SUSPEND live_view, ...]
* [RESUME live_view, ...]
* [REFRESH live_view, ...]
* [MODIFY live_view, ...]
*/
class ParserAlterQuery : public IParserBase
@ -35,6 +44,12 @@ class ParserAlterCommandList : public IParserBase
protected:
const char * getName() const { return "a list of ALTER commands"; }
bool parseImpl(Pos & pos, ASTPtr & node, Expected & expected);
public:
bool is_live_view;
bool is_live_channel;
ParserAlterCommandList(bool is_live_view_ = false, bool is_live_channel_ = false) : is_live_view(is_live_view_), is_live_channel(is_live_channel_) {}
};
@ -43,6 +58,12 @@ class ParserAlterCommand : public IParserBase
protected:
const char * getName() const { return "ALTER command"; }
bool parseImpl(Pos & pos, ASTPtr & node, Expected & expected);
public:
bool is_live_view;
bool is_live_channel;
ParserAlterCommand(bool is_live_view_ = false, bool is_live_channel_ = false) : is_live_view(is_live_view_), is_live_channel(is_live_channel_) {}
};

90
dbms/src/Parsers/ParserCreateQuery.cpp
View File

@ -94,6 +94,12 @@ bool ParserColumnDeclarationList::parseImpl(Pos & pos, ASTPtr & node, Expected &
.parse(pos, node, expected);
}
bool ParserNameList::parseImpl(Pos & pos, ASTPtr & node, Expected & expected)
{
return ParserList(std::make_unique<ParserCompoundIdentifier>(), std::make_unique<ParserToken>(TokenType::Comma), false)
.parse(pos, node, expected);
}
bool ParserIndexDeclaration::parseImpl(Pos & pos, ASTPtr & node, Expected & expected)
{
ParserKeyword s_type("TYPE");
@ -309,7 +315,10 @@ bool ParserCreateQuery::parseImpl(Pos & pos, ASTPtr & node, Expected & expected)
ParserKeyword s_if_not_exists("IF NOT EXISTS");
ParserKeyword s_as("AS");
ParserKeyword s_view("VIEW");
ParserKeyword s_with("WITH");
ParserKeyword s_materialized("MATERIALIZED");
ParserKeyword s_live("LIVE");
ParserKeyword s_channel("CHANNEL");
ParserKeyword s_populate("POPULATE");
ParserKeyword s_or_replace("OR REPLACE");
ParserToken s_dot(TokenType::Dot);
@ -320,6 +329,7 @@ bool ParserCreateQuery::parseImpl(Pos & pos, ASTPtr & node, Expected & expected)
ParserColumnsOrIndicesDeclarationList columns_or_indices_p;
ParserSelectWithUnionQuery select_p;
ParserFunction table_function_p;
ParserNameList names_p;
ASTPtr database;
ASTPtr table;
@ -331,11 +341,15 @@ bool ParserCreateQuery::parseImpl(Pos & pos, ASTPtr & node, Expected & expected)
ASTPtr as_table;
ASTPtr as_table_function;
ASTPtr select;
ASTPtr tables;
String cluster_str;
bool attach = false;
bool if_not_exists = false;
bool is_view = false;
bool is_materialized_view = false;
bool is_live_view = false;
bool is_live_channel = false;
bool is_populate = false;
bool is_temporary = false;
bool replace_view = false;
@ -431,6 +445,79 @@ bool ParserCreateQuery::parseImpl(Pos & pos, ASTPtr & node, Expected & expected)
}
}
}
else if (s_live.ignore(pos, expected))
{
if (s_channel.ignore(pos, expected))
is_live_channel = true;
else if (s_view.ignore(pos, expected))
is_live_view = true;
else
return false;
if (s_if_not_exists.ignore(pos, expected))
if_not_exists = true;
if (!name_p.parse(pos, table, expected))
return false;
if (s_dot.ignore(pos, expected))
{
database = table;
if (!name_p.parse(pos, table, expected))
return false;
}
if (ParserKeyword{"ON"}.ignore(pos, expected))
{
if (!ASTQueryWithOnCluster::parse(pos, cluster_str, expected))
return false;
}
if (!is_live_channel)
{
// TO [db.]table
if (ParserKeyword{"TO"}.ignore(pos, expected))
{
if (!name_p.parse(pos, to_table, expected))
return false;
if (s_dot.ignore(pos, expected))
{
to_database = to_table;
if (!name_p.parse(pos, to_table, expected))
return false;
}
}
}
/// Optional - a list of columns can be specified. It must fully comply with SELECT.
if (s_lparen.ignore(pos, expected))
{
if (!columns_or_indices_p.parse(pos, columns_list, expected))
return false;
if (!s_rparen.ignore(pos, expected))
return false;
}
if (is_live_channel)
{
if (s_with.ignore(pos, expected))
{
if (!names_p.parse(pos, tables, expected))
return false;
}
}
else
{
/// AS SELECT ...
if (!s_as.ignore(pos, expected))
return false;
if (!select_p.parse(pos, select, expected))
return false;
}
}
else if (is_temporary)
return false;
else if (s_database.ignore(pos, expected))
@ -538,6 +625,8 @@ bool ParserCreateQuery::parseImpl(Pos & pos, ASTPtr & node, Expected & expected)
query->if_not_exists = if_not_exists;
query->is_view = is_view;
query->is_materialized_view = is_materialized_view;
query->is_live_view = is_live_view;
query->is_live_channel = is_live_channel;
query->is_populate = is_populate;
query->temporary = is_temporary;
query->replace_view = replace_view;
@ -551,6 +640,7 @@ bool ParserCreateQuery::parseImpl(Pos & pos, ASTPtr & node, Expected & expected)
query->set(query->columns_list, columns_list);
query->set(query->storage, storage);
query->set(query->tables, tables);
tryGetIdentifierNameInto(as_database, query->as_database);
tryGetIdentifierNameInto(as_table, query->as_table);

10
dbms/src/Parsers/ParserCreateQuery.h
View File

@ -91,6 +91,14 @@ protected:
bool parseImpl(Pos & pos, ASTPtr & node, Expected & expected);
};
/** List of table names. */
class ParserNameList : public IParserBase
{
protected:
const char * getName() const { return "name list"; }
bool parseImpl(Pos & pos, ASTPtr & node, Expected & expected);
};
template <typename NameParser>
class IParserColumnDeclaration : public IParserBase
@ -300,7 +308,7 @@ protected:
* CREATE|ATTACH DATABASE db [ENGINE = engine]
*
* Or:
* CREATE [OR REPLACE]|ATTACH [MATERIALIZED] VIEW [IF NOT EXISTS] [db.]name [TO [db.]name] [ENGINE = engine] [POPULATE] AS SELECT ...
* CREATE[OR REPLACE]|ATTACH [[MATERIALIZED] VIEW] | [[TEMPORARY] LIVE [CHANNEL] | [VIEW]] [IF NOT EXISTS] [db.]name [TO [db.]name] [ENGINE = engine] [POPULATE] AS SELECT ...
*/
class ParserCreateQuery : public IParserBase
{

5
dbms/src/Parsers/ParserQueryWithOutput.cpp
View File

@ -11,6 +11,7 @@
#include <Parsers/ParserDropQuery.h>
#include <Parsers/ParserKillQueryQuery.h>
#include <Parsers/ParserOptimizeQuery.h>
#include <Parsers/ParserWatchQuery.h>
#include <Parsers/ParserSetQuery.h>
#include <Parsers/ASTExplainQuery.h>
@ -32,6 +33,7 @@ bool ParserQueryWithOutput::parseImpl(Pos & pos, ASTPtr & node, Expected & expec
ParserCheckQuery check_p;
ParserOptimizeQuery optimize_p;
ParserKillQueryQuery kill_query_p;
ParserWatchQuery watch_p;
ASTPtr query;
@ -57,7 +59,8 @@ bool ParserQueryWithOutput::parseImpl(Pos & pos, ASTPtr & node, Expected & expec
|| drop_p.parse(pos, query, expected)
|| check_p.parse(pos, query, expected)
|| kill_query_p.parse(pos, query, expected)
|| optimize_p.parse(pos, query, expected);
|| optimize_p.parse(pos, query, expected)
|| watch_p.parse(pos, query, expected);
if (!parsed)
return false;

77
dbms/src/Parsers/ParserWatchQuery.cpp Normal file
View File

@ -0,0 +1,77 @@
/* Copyright (c) 2018 BlackBerry Limited
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License. */
#include <Parsers/ASTLiteral.h>
#include <Parsers/ASTIdentifier.h>
#include <Parsers/ASTWatchQuery.h>
#include <Parsers/CommonParsers.h>
#include <Parsers/ParserWatchQuery.h>
#include <Parsers/ExpressionElementParsers.h>
namespace DB
{
bool ParserWatchQuery::parseImpl(Pos & pos, ASTPtr & node, Expected & expected)
{
ParserKeyword s_watch("WATCH");
ParserToken s_dot(TokenType::Dot);
ParserIdentifier name_p;
ParserKeyword s_events("EVENTS");
ParserKeyword s_limit("LIMIT");
ASTPtr database;
ASTPtr table;
auto query = std::make_shared<ASTWatchQuery>();
if (!s_watch.ignore(pos, expected))
{
return false;
}
if (!name_p.parse(pos, table, expected))
return false;
if (s_dot.ignore(pos, expected))
{
database = table;
if (!name_p.parse(pos, table, expected))
return false;
}
/// EVENTS
if (s_events.ignore(pos, expected))
{
query->is_watch_events = true;
}
/// LIMIT length
if (s_limit.ignore(pos, expected))
{
ParserNumber num;
if (!num.parse(pos, query->limit_length, expected))
return false;
}
if (database)
query->database = getIdentifierName(database);
if (table)
query->table = getIdentifierName(table);
node = query;
return true;
}
}

30
dbms/src/Parsers/ParserWatchQuery.h Normal file
View File

@ -0,0 +1,30 @@
/* Copyright (c) 2018 BlackBerry Limited
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License. */
#pragma once
#include <Parsers/IParserBase.h>
namespace DB
{
/** Query like this:
* WATCH [db.]table EVENTS
*/
class ParserWatchQuery : public IParserBase
{
protected:
const char * getName() const { return "WATCH query"; }
bool parseImpl(Pos & pos, ASTPtr & node, Expected & expected);
};
}

6
dbms/src/Processors/Formats/IRowOutputFormat.cpp
View File

@ -20,6 +20,9 @@ void IRowOutputFormat::consume(DB::Chunk chunk)
first_row = false;
write(columns, row);
if (write_single_row_callback)
write_single_row_callback();
}
}
@ -96,6 +99,3 @@ void IRowOutputFormat::writeTotals(const DB::Columns & columns, size_t row_num)
}
}

13
dbms/src/Processors/Formats/IRowOutputFormat.h
View File

@ -1,8 +1,10 @@
#pragma once
#include <string>
#include <Formats/FormatFactory.h>
#include <Processors/Formats/IOutputFormat.h>
#include <string>
namespace DB
{
@ -22,8 +24,8 @@ protected:
void finalize() override;
public:
IRowOutputFormat(const Block & header, WriteBuffer & out_)
: IOutputFormat(header, out_), types(header.getDataTypes())
IRowOutputFormat(const Block & header, WriteBuffer & out_, FormatFactory::WriteCallback callback)
: IOutputFormat(header, out_), types(header.getDataTypes()), write_single_row_callback(callback)
{
}
@ -57,6 +59,9 @@ private:
bool prefix_written = false;
bool suffix_written = false;
// Callback used to indicate that another row is written.
FormatFactory::WriteCallback write_single_row_callback;
void writePrefixIfNot()
{
if (!prefix_written)
@ -76,5 +81,3 @@ private:
};
}

10
dbms/src/Processors/Formats/Impl/BinaryRowOutputFormat.cpp
View File

@ -9,8 +9,8 @@
namespace DB
{
BinaryRowOutputFormat::BinaryRowOutputFormat(WriteBuffer & out_, const Block & header, bool with_names_, bool with_types_)
: IRowOutputFormat(header, out_), with_names(with_names_), with_types(with_types_)
BinaryRowOutputFormat::BinaryRowOutputFormat(WriteBuffer & out_, const Block & header, bool with_names_, bool with_types_, FormatFactory::WriteCallback callback)
: IRowOutputFormat(header, out_, callback), with_names(with_names_), with_types(with_types_)
{
}
@ -53,18 +53,20 @@ void registerOutputFormatProcessorRowBinary(FormatFactory & factory)
WriteBuffer & buf,
const Block & sample,
const Context &,
FormatFactory::WriteCallback callback,
const FormatSettings &)
{
return std::make_shared<BinaryRowOutputFormat>(buf, sample, false, false);
return std::make_shared<BinaryRowOutputFormat>(buf, sample, false, false, callback);
});
factory.registerOutputFormatProcessor("RowBinaryWithNamesAndTypes", [](
WriteBuffer & buf,
const Block & sample,
const Context &,
FormatFactory::WriteCallback callback,
const FormatSettings &)
{
return std::make_shared<BinaryRowOutputFormat>(buf, sample, true, true);
return std::make_shared<BinaryRowOutputFormat>(buf, sample, true, true, callback);
});
}

3
dbms/src/Processors/Formats/Impl/BinaryRowOutputFormat.h
View File

@ -17,7 +17,7 @@ class WriteBuffer;
class BinaryRowOutputFormat: public IRowOutputFormat
{
public:
BinaryRowOutputFormat(WriteBuffer & out_, const Block & header, bool with_names_, bool with_types_);
BinaryRowOutputFormat(WriteBuffer & out_, const Block & header, bool with_names_, bool with_types_, FormatFactory::WriteCallback callback);
String getName() const override { return "BinaryRowOutputFormat"; }
@ -32,4 +32,3 @@ protected:
};
}

10
dbms/src/Processors/Formats/Impl/CSVRowInputFormat.cpp
View File

@ -349,10 +349,9 @@ bool OPTIMIZE(1) CSVRowInputFormat::parseRowAndPrintDiagnosticInfo(MutableColumn
const auto & current_column_type = data_types[table_column];
const bool is_last_file_column =
file_column + 1 == column_indexes_for_input_fields.size();
const bool at_delimiter = *in.position() == delimiter;
const bool at_delimiter = !in.eof() && *in.position() == delimiter;
const bool at_last_column_line_end = is_last_file_column
&& (*in.position() == '\n' || *in.position() == '\r'
|| in.eof());
&& (in.eof() || *in.position() == '\n' || *in.position() == '\r');
auto & header = getPort().getHeader();
out << "Column " << file_column << ", " << std::string((file_column < 10 ? 2 : file_column < 100 ? 1 : 0), ' ')
@ -516,10 +515,9 @@ void CSVRowInputFormat::updateDiagnosticInfo()
bool CSVRowInputFormat::readField(IColumn & column, const DataTypePtr & type, bool is_last_file_column, size_t column_idx)
{
const bool at_delimiter = *in.position() == format_settings.csv.delimiter;
const bool at_delimiter = !in.eof() && *in.position() == format_settings.csv.delimiter;
const bool at_last_column_line_end = is_last_file_column
&& (*in.position() == '\n' || *in.position() == '\r'
|| in.eof());
&& (in.eof() || *in.position() == '\n' || *in.position() == '\r');
if (format_settings.csv.empty_as_default
&& (at_delimiter || at_last_column_line_end))

7
dbms/src/Processors/Formats/Impl/CSVRowOutputFormat.cpp
View File

@ -8,8 +8,8 @@ namespace DB
{
CSVRowOutputFormat::CSVRowOutputFormat(WriteBuffer & out_, const Block & header_, bool with_names_, const FormatSettings & format_settings_)
: IRowOutputFormat(header_, out_), with_names(with_names_), format_settings(format_settings_)
CSVRowOutputFormat::CSVRowOutputFormat(WriteBuffer & out_, const Block & header_, bool with_names_, FormatFactory::WriteCallback callback, const FormatSettings & format_settings_)
: IRowOutputFormat(header_, out_, callback), with_names(with_names_), format_settings(format_settings_)
{
auto & sample = getPort(PortKind::Main).getHeader();
size_t columns = sample.columns();
@ -77,9 +77,10 @@ void registerOutputFormatProcessorCSV(FormatFactory & factory)
WriteBuffer & buf,
const Block & sample,
const Context &,
FormatFactory::WriteCallback callback,
const FormatSettings & format_settings)
{
return std::make_shared<CSVRowOutputFormat>(buf, sample, with_names, format_settings);
return std::make_shared<CSVRowOutputFormat>(buf, sample, with_names, callback, format_settings);
});
}
}

3
dbms/src/Processors/Formats/Impl/CSVRowOutputFormat.h
View File

@ -20,7 +20,7 @@ public:
/** with_names - output in the first line a header with column names
* with_types - output in the next line header with the names of the types
*/
CSVRowOutputFormat(WriteBuffer & out_, const Block & header_, bool with_names_, const FormatSettings & format_settings_);
CSVRowOutputFormat(WriteBuffer & out_, const Block & header_, bool with_names_, FormatFactory::WriteCallback callback, const FormatSettings & format_settings_);
String getName() const override { return "CSVRowOutputFormat"; }
@ -45,4 +45,3 @@ protected:
};
}

7
dbms/src/Processors/Formats/Impl/JSONCompactRowOutputFormat.cpp
View File

@ -8,8 +8,8 @@ namespace DB
{
JSONCompactRowOutputFormat::JSONCompactRowOutputFormat(
WriteBuffer & out_, const Block & header, const FormatSettings & settings_)
: JSONRowOutputFormat(out_, header, settings_)
WriteBuffer & out_, const Block & header, FormatFactory::WriteCallback callback, const FormatSettings & settings_)
: JSONRowOutputFormat(out_, header, callback, settings_)
{
}
@ -81,9 +81,10 @@ void registerOutputFormatProcessorJSONCompact(FormatFactory & factory)
WriteBuffer & buf,
const Block & sample,
const Context &,
FormatFactory::WriteCallback callback,
const FormatSettings & format_settings)
{
return std::make_shared<JSONCompactRowOutputFormat>(buf, sample, format_settings);
return std::make_shared<JSONCompactRowOutputFormat>(buf, sample, callback, format_settings);
});
}

2
dbms/src/Processors/Formats/Impl/JSONCompactRowOutputFormat.h
View File

@ -16,7 +16,7 @@ struct FormatSettings;
class JSONCompactRowOutputFormat : public JSONRowOutputFormat
{
public:
JSONCompactRowOutputFormat(WriteBuffer & out_, const Block & header, const FormatSettings & settings_);
JSONCompactRowOutputFormat(WriteBuffer & out_, const Block & header, FormatFactory::WriteCallback callback, const FormatSettings & settings_);
String getName() const override { return "JSONCompactRowOutputFormat"; }

Some files were not shown because too many files have changed in this diff Show More