ClickHouse

1
#include "ODBCSink.h"
2

3
#include <IO/WriteBufferFromString.h>
4
#include <Interpreters/Context.h>
5
#include <Processors/Formats/IOutputFormat.h>
6
#include <Parsers/getInsertQuery.h>
7

8

9
namespace DB
10
{
11

12

13
ODBCSink::ODBCSink(
14
    nanodbc::ConnectionHolderPtr connection_holder_,
15
    const std::string & remote_database_name_,
16
    const std::string & remote_table_name_,
17
    const Block & sample_block_,
18
    ContextPtr local_context_,
19
    IdentifierQuotingStyle quoting_)
20
    : ISink(sample_block_)
21
    , log(getLogger("ODBCSink"))
22
    , connection_holder(std::move(connection_holder_))
23
    , db_name(remote_database_name_)
24
    , table_name(remote_table_name_)
25
    , sample_block(sample_block_)
26
    , local_context(local_context_)
27
    , quoting(quoting_)
28
{
29
    description.init(sample_block);
30
}
31

32

33
void ODBCSink::consume(Chunk chunk)
34
{
35
    auto block = getPort().getHeader().cloneWithColumns(chunk.detachColumns());
36
    WriteBufferFromOwnString values_buf;
37
    auto writer = local_context->getOutputFormat("Values", values_buf, sample_block);
38
    writer->write(block);
39

40
    std::string query = getInsertQuery(db_name, table_name, block.getColumnsWithTypeAndName(), quoting) + values_buf.str();
41
    execute<void>(connection_holder,
42
        [&](nanodbc::connection & connection) { execute(connection, query); });
43
}
44

45
}
46