1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
|
#include <Processors/Transforms/SquashingChunksTransform.h>
namespace DB
{
SquashingChunksTransform::SquashingChunksTransform(
const Block & header, size_t min_block_size_rows, size_t min_block_size_bytes)
: ExceptionKeepingTransform(header, header, false)
, squashing(min_block_size_rows, min_block_size_bytes)
{
}
void SquashingChunksTransform::onConsume(Chunk chunk)
{
if (auto block = squashing.add(getInputPort().getHeader().cloneWithColumns(chunk.detachColumns())))
{
cur_chunk.setColumns(block.getColumns(), block.rows());
}
}
SquashingChunksTransform::GenerateResult SquashingChunksTransform::onGenerate()
{
GenerateResult res;
res.chunk = std::move(cur_chunk);
res.is_done = true;
return res;
}
void SquashingChunksTransform::onFinish()
{
auto block = squashing.add({});
finish_chunk.setColumns(block.getColumns(), block.rows());
}
void SquashingChunksTransform::work()
{
if (stage == Stage::Exception)
{
data.chunk.clear();
ready_input = false;
return;
}
ExceptionKeepingTransform::work();
if (finish_chunk)
{
data.chunk = std::move(finish_chunk);
ready_output = true;
}
}
SimpleSquashingChunksTransform::SimpleSquashingChunksTransform(
const Block & header, size_t min_block_size_rows, size_t min_block_size_bytes)
: ISimpleTransform(header, header, true), squashing(min_block_size_rows, min_block_size_bytes)
{
}
void SimpleSquashingChunksTransform::transform(Chunk & chunk)
{
if (!finished)
{
if (auto block = squashing.add(getInputPort().getHeader().cloneWithColumns(chunk.detachColumns())))
chunk.setColumns(block.getColumns(), block.rows());
}
else
{
auto block = squashing.add({});
chunk.setColumns(block.getColumns(), block.rows());
}
}
IProcessor::Status SimpleSquashingChunksTransform::prepare()
{
if (!finished && input.isFinished())
{
finished = true;
return Status::Ready;
}
return ISimpleTransform::prepare();
}
}
|
