2019-03-26 18:28:37 +00:00
|
|
|
#include <Processors/Sources/SourceFromInputStream.h>
|
2019-04-05 15:45:46 +00:00
|
|
|
#include <Processors/Transforms/AggregatingTransform.h>
|
|
|
|
#include <DataTypes/DataTypeAggregateFunction.h>
|
2019-03-26 18:28:37 +00:00
|
|
|
|
|
|
|
namespace DB
|
|
|
|
{
|
|
|
|
|
|
|
|
SourceFromInputStream::SourceFromInputStream(Block header, BlockInputStreamPtr stream)
|
|
|
|
: ISource(std::move(header)), stream(std::move(stream))
|
|
|
|
{
|
2019-04-05 15:45:46 +00:00
|
|
|
auto & sample = getPort().getHeader();
|
|
|
|
for (auto & type : sample.getDataTypes())
|
|
|
|
if (typeid_cast<const DataTypeAggregateFunction *>(type.get()))
|
|
|
|
has_aggregate_functions = true;
|
2019-03-26 18:28:37 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
Chunk SourceFromInputStream::generate()
|
|
|
|
{
|
2019-04-05 11:27:08 +00:00
|
|
|
if (stream_finished)
|
2019-03-26 18:28:37 +00:00
|
|
|
return {};
|
|
|
|
|
|
|
|
if (!initialized)
|
|
|
|
{
|
|
|
|
stream->readPrefix();
|
|
|
|
initialized = true;
|
|
|
|
}
|
|
|
|
|
|
|
|
auto block = stream->read();
|
|
|
|
if (!block)
|
|
|
|
{
|
|
|
|
stream->readSuffix();
|
2019-04-05 11:27:08 +00:00
|
|
|
stream_finished = true;
|
2019-03-26 18:28:37 +00:00
|
|
|
return {};
|
|
|
|
}
|
|
|
|
|
|
|
|
assertBlocksHaveEqualStructure(getPort().getHeader(), block, "SourceFromInputStream");
|
|
|
|
|
|
|
|
UInt64 num_rows = block.rows();
|
2019-04-05 15:45:46 +00:00
|
|
|
Chunk chunk(block.getColumns(), num_rows);
|
|
|
|
|
|
|
|
if (has_aggregate_functions)
|
|
|
|
{
|
|
|
|
auto info = std::make_shared<AggregatedChunkInfo>();
|
|
|
|
info->bucket_num = block.info.bucket_num;
|
|
|
|
info->is_overflows = block.info.is_overflows;
|
|
|
|
chunk.setChunkInfo(std::move(info));
|
|
|
|
}
|
|
|
|
|
|
|
|
return chunk;
|
2019-03-26 18:28:37 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
}
|