ClickHouse

1
#include <Processors/Port.h>
2
#include <Processors/IProcessor.h>
3

4
namespace DB
5
{
6
namespace ErrorCodes
7
{
8
    extern const int LOGICAL_ERROR;
9
}
10

11
void connect(OutputPort & output, InputPort & input, bool reconnect)
12
{
13
    if (!reconnect && input.state)
14
        throw Exception(ErrorCodes::LOGICAL_ERROR, "Port is already connected, (header: [{}])", input.header.dumpStructure());
15

16
    if (!reconnect && output.state)
17
        throw Exception(ErrorCodes::LOGICAL_ERROR, "Port is already connected, (header: [{}])", output.header.dumpStructure());
18

19
    auto out_name = output.processor ? output.getProcessor().getName() : "null";
20
    auto in_name = input.processor ? input.getProcessor().getName() : "null";
21

22
    assertCompatibleHeader(output.getHeader(), input.getHeader(), fmt::format("function connect between {} and {}", out_name, in_name));
23

24
    input.output_port = &output;
25
    output.input_port = &input;
26
    input.state = std::make_shared<Port::State>();
27
    output.state = input.state;
28
}
29

30
}
31