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

Fix some more tests.

Browse Source
This commit is contained in:
Nikolai Kochetov 2022-06-01 15:21:47 +00:00
parent e401ab8169
commit 6e924cdc77
5 changed files with 19 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
programs/copier/ClusterCopier.cpp
View File

@ -8,7 +8,7 @@
#include <Common/setThreadName.h> #include <Common/setThreadName.h>
#include <IO/ConnectionTimeoutsContext.h> #include <IO/ConnectionTimeoutsContext.h>
#include <Interpreters/InterpreterInsertQuery.h> #include <Interpreters/InterpreterInsertQuery.h>
#include <Interpreters/InterpreterSelectQuery.h> #include <Interpreters/InterpreterSelectWithUnionQuery.h>
#include <Parsers/ASTFunction.h> #include <Parsers/ASTFunction.h>
#include <Processors/Transforms/ExpressionTransform.h> #include <Processors/Transforms/ExpressionTransform.h>
#include <QueryPipeline/QueryPipelineBuilder.h> #include <QueryPipeline/QueryPipelineBuilder.h>
@ -1455,7 +1455,7 @@ TaskStatus ClusterCopier::processPartitionPieceTaskImpl(
local_context->setSettings(task_cluster->settings_pull); local_context->setSettings(task_cluster->settings_pull);
local_context->setSetting("skip_unavailable_shards", true); local_context->setSetting("skip_unavailable_shards", true);
InterpreterSelectQuery select(query_select_ast, local_context, SelectQueryOptions{}); InterpreterSelectWithUnionQuery select(query_select_ast, local_context, SelectQueryOptions{});
QueryPlan plan; QueryPlan plan;
select.buildQueryPlan(plan); select.buildQueryPlan(plan);
auto builder = std::move(*plan.buildQueryPipeline( auto builder = std::move(*plan.buildQueryPipeline(
@ -1545,7 +1545,7 @@ TaskStatus ClusterCopier::processPartitionPieceTaskImpl(
{ {
BlockIO io_insert = InterpreterFactory::get(query_insert_ast, context_insert)->execute(); BlockIO io_insert = InterpreterFactory::get(query_insert_ast, context_insert)->execute();
InterpreterSelectQuery select(query_select_ast, context_select, SelectQueryOptions{}); InterpreterSelectWithUnionQuery select(query_select_ast, context_select, SelectQueryOptions{});
QueryPlan plan; QueryPlan plan;
select.buildQueryPlan(plan); select.buildQueryPlan(plan);
auto builder = std::move(*plan.buildQueryPipeline( auto builder = std::move(*plan.buildQueryPipeline(
@ -1875,7 +1875,7 @@ std::set<String> ClusterCopier::getShardPartitions(const ConnectionTimeouts & ti
auto local_context = Context::createCopy(context); auto local_context = Context::createCopy(context);
local_context->setSettings(task_cluster->settings_pull); local_context->setSettings(task_cluster->settings_pull);
InterpreterSelectQuery select(query_ast, local_context, SelectQueryOptions{}); InterpreterSelectWithUnionQuery select(query_ast, local_context, SelectQueryOptions{});
QueryPlan plan; QueryPlan plan;
select.buildQueryPlan(plan); select.buildQueryPlan(plan);
auto builder = std::move(*plan.buildQueryPipeline( auto builder = std::move(*plan.buildQueryPipeline(

2
src/Processors/ISource.h
View File

@ -40,7 +40,7 @@ public:
void setStorageLimits(const std::shared_ptr<const StorageLimitsList> & storage_limits_) override; void setStorageLimits(const std::shared_ptr<const StorageLimitsList> & storage_limits_) override;
/// Default implementation for all the sources. /// Default implementation for all the sources.
std::optional<ReadProgress> getReadProgress() final; std::optional<ReadProgress> getReadProgress() override;
void addTotalRowsApprox(size_t value) { read_progress.total_rows_approx += value; } void addTotalRowsApprox(size_t value) { read_progress.total_rows_approx += value; }
}; };

2
src/Processors/Sources/NullSource.h
View File

@ -11,6 +11,8 @@ public:
explicit NullSource(Block header) : ISource(std::move(header)) {} explicit NullSource(Block header) : ISource(std::move(header)) {}
String getName() const override { return "NullSource"; } String getName() const override { return "NullSource"; }
std::optional<ReadProgress> getReadProgress() override { return {}; }
protected: protected:
Chunk generate() override { return Chunk(); } Chunk generate() override { return Chunk(); }
}; };

4
src/Processors/Sources/RemoteSource.h
View File

@ -61,6 +61,8 @@ public:
String getName() const override { return "RemoteTotals"; } String getName() const override { return "RemoteTotals"; }
std::optional<ReadProgress> getReadProgress() override { return {}; }
protected: protected:
Chunk generate() override; Chunk generate() override;
@ -77,6 +79,8 @@ public:
String getName() const override { return "RemoteExtremes"; } String getName() const override { return "RemoteExtremes"; }
std::optional<ReadProgress> getReadProgress() override { return {}; }
protected: protected:
Chunk generate() override; Chunk generate() override;

8
src/Storages/ReadFinalForExternalReplicaStorage.cpp
View File

@ -13,6 +13,7 @@
#include <Processors/QueryPlan/FilterStep.h> #include <Processors/QueryPlan/FilterStep.h>
#include <Processors/QueryPlan/QueryPlan.h> #include <Processors/QueryPlan/QueryPlan.h>
#include <Interpreters/Context.h> #include <Interpreters/Context.h>
#include <Interpreters/InterpreterSelectQuery.h>
namespace DB namespace DB
@ -59,6 +60,13 @@ void readFinalFromNestedStorage(
auto nested_snapshot = nested_storage->getStorageSnapshot(nested_metadata, context); auto nested_snapshot = nested_storage->getStorageSnapshot(nested_metadata, context);
nested_storage->read(query_plan, require_columns_name, nested_snapshot, query_info, context, processed_stage, max_block_size, num_streams); nested_storage->read(query_plan, require_columns_name, nested_snapshot, query_info, context, processed_stage, max_block_size, num_streams);
if (!query_plan.isInitialized())
{
InterpreterSelectQuery::addEmptySourceToQueryPlan(query_plan, nested_header, query_info, context);
return;
}
query_plan.addTableLock(lock); query_plan.addTableLock(lock);
query_plan.addStorageHolder(nested_storage); query_plan.addStorageHolder(nested_storage);