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

Take header from first block in Squashing stream

Browse Source
This commit is contained in:
alesapin 2018-09-19 20:39:04 +03:00
parent e1c8045055
commit 2b3e29f607
2 changed files with 15 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -4,14 +4,22 @@
namespace DB
{
namespace ErrorCodes
{
extern const int LOGICAL_ERROR;
}
SquashingBlockOutputStream::SquashingBlockOutputStream(BlockOutputStreamPtr & dst, size_t min_block_size_rows, size_t min_block_size_bytes)
: output(dst), header(output->getHeader()), transform(min_block_size_rows, min_block_size_bytes)
: output(dst), transform(min_block_size_rows, min_block_size_bytes)
{
}
void SquashingBlockOutputStream::write(const Block & block)
{
if (!header)
header = block.cloneEmpty();
SquashingTransform::Result result = transform.add(Block(block).mutateColumns());
if (result.ready)
output->write(header.cloneWithColumns(std::move(result.columns)));
@ -25,6 +33,9 @@ void SquashingBlockOutputStream::finalize()
all_written = true;
if (!header)
throw Exception("writeSuffix called without writing data.", ErrorCodes::LOGICAL_ERROR);
SquashingTransform::Result result = transform.add({});
if (result.ready && !result.columns.empty())
output->write(header.cloneWithColumns(std::move(result.columns)));

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

@ -96,6 +96,9 @@ BlockIO InterpreterInsertQuery::execute()
out = std::make_shared<PushingToViewsBlockOutputStream>(query.database, query.table, table, context, query_ptr, query.no_destination);
out = std::make_shared<AddingDefaultBlockOutputStream>(
out, getSampleBlock(query, table), required_columns, table->getColumns().defaults, context);
/// Do not squash blocks if it is a sync INSERT into Distributed, since it lead to double bufferization on client and server side.
/// Client-side bufferization might cause excessive timeouts (especially in case of big blocks).
if (!(context.getSettingsRef().insert_distributed_sync && table->isRemote()))
@ -104,9 +107,6 @@ BlockIO InterpreterInsertQuery::execute()
out, context.getSettingsRef().min_insert_block_size_rows, context.getSettingsRef().min_insert_block_size_bytes);
}
out = std::make_shared<AddingDefaultBlockOutputStream>(
out, getSampleBlock(query, table), required_columns, table->getColumns().defaults, context);
auto out_wrapper = std::make_shared<CountingBlockOutputStream>(out);
out_wrapper->setProcessListElement(context.getProcessListElement());
out = std::move(out_wrapper);