blob: 629529cdffa0e4fee1eb5de347e9eefc12780a72 (
plain) (
blame)
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
|
#pragma once
#include <Processors/IProcessor.h>
namespace DB
{
/** Has one input and one output.
* Simply pull a block from input, transform it, and push it to output.
*/
class ISimpleTransform : public IProcessor
{
protected:
InputPort & input;
OutputPort & output;
Port::Data input_data;
Port::Data output_data;
bool has_input = false;
bool has_output = false;
bool no_more_data_needed = false;
const bool skip_empty_chunks;
/// Set input port NotNeeded after chunk was pulled.
/// Input port will become needed again only after data was transformed.
/// This allows to escape caching chunks in input port, which can lead to uneven data distribution.
bool set_input_not_needed_after_read = true;
virtual void transform(Chunk & input_chunk, Chunk & output_chunk)
{
transform(input_chunk);
output_chunk.swap(input_chunk);
}
virtual bool needInputData() const { return true; }
void stopReading() { no_more_data_needed = true; }
public:
ISimpleTransform(Block input_header_, Block output_header_, bool skip_empty_chunks_);
virtual void transform(Chunk &) = 0;
Status prepare() override;
void work() override;
InputPort & getInputPort() { return input; }
OutputPort & getOutputPort() { return output; }
void setInputNotNeededAfterRead(bool value) { set_input_not_needed_after_read = value; }
};
}
|
