CLICKHOUSE-3878: Add odbc-bridge first version

dbms/programs/CMakeLists.txt

@@ -13,6 +13,10 @@ option (ENABLE_CLICKHOUSE_COMPRESSOR "Enable clickhouse-compressor" ${ENABLE_CLI
 option (ENABLE_CLICKHOUSE_COPIER "Enable clickhouse-copier" ${ENABLE_CLICKHOUSE_ALL})
 option (ENABLE_CLICKHOUSE_FORMAT "Enable clickhouse-format" ${ENABLE_CLICKHOUSE_ALL})
 option (ENABLE_CLICKHOUSE_OBFUSCATOR "Enable clickhouse-obfuscator" ${ENABLE_CLICKHOUSE_ALL})
+option (ENABLE_CLICKHOUSE_ODBC_BRIDGE "Enable clickhouse-odbc-bridge" ${ENABLE_CLICKHOUSE_ALL})
+
+MESSAGE(STATUS, "ENABLED_ODBC")
+MESSAGE(STATUS, ${ENABLE_CLICKHOUSE_ODBC_BRIDGE})
 
 configure_file (config_tools.h.in ${CMAKE_CURRENT_BINARY_DIR}/config_tools.h)
 
@@ -27,10 +31,11 @@ add_subdirectory (copier)
 add_subdirectory (format)
 add_subdirectory (clang)
 add_subdirectory (obfuscator)
+add_subdirectory (odbc-bridge)
 
 if (CLICKHOUSE_SPLIT_BINARY)
     set (CLICKHOUSE_ALL_TARGETS clickhouse-server clickhouse-client clickhouse-local clickhouse-benchmark clickhouse-performance-test
-            clickhouse-extract-from-config clickhouse-format clickhouse-copier)
+            clickhouse-extract-from-config clickhouse-format clickhouse-copier clickhouse-odbc-bridge)
 
     if (USE_EMBEDDED_COMPILER)
         list (APPEND CLICKHOUSE_ALL_TARGETS clickhouse-clang clickhouse-lld)
@@ -83,6 +88,9 @@ else ()
     if (USE_EMBEDDED_COMPILER)
         target_link_libraries (clickhouse clickhouse-compiler-lib)
     endif ()
+    if (ENABLE_CLICKHOUSE_ODBC_BRIDGE)
+        target_link_libraries (clickhouse clickhouse-odbc-bridge-lib)
+    endif()
 
     set (CLICKHOUSE_BUNDLE)
     if (ENABLE_CLICKHOUSE_SERVER)
@@ -135,6 +143,12 @@ else ()
         install (FILES ${CMAKE_CURRENT_BINARY_DIR}/clickhouse-obfuscator DESTINATION ${CMAKE_INSTALL_BINDIR} COMPONENT clickhouse)
         list(APPEND CLICKHOUSE_BUNDLE clickhouse-obfuscator)
     endif ()
+    if (ENABLE_CLICKHOUSE_ODBC_BRIDGE)
+        add_custom_target (clickhouse-odbc-bridge ALL COMMAND ${CMAKE_COMMAND} -E create_symlink clickhouse clickhouse-odbc-bridge DEPENDS clickhouse)
+        install (FILES ${CMAKE_CURRENT_BINARY_DIR}/clickhouse-odbc-bridge DESTINATION ${CMAKE_INSTALL_BINDIR} COMPONENT clickhouse)
+        list(APPEND CLICKHOUSE_BUNDLE clickhouse-odbc-bridge)
+    endif ()
+
 
     # install always because depian package want this files:
     add_custom_target (clickhouse-clang ALL COMMAND ${CMAKE_COMMAND} -E create_symlink clickhouse clickhouse-clang DEPENDS clickhouse)

dbms/programs/main.cpp

@@ -56,6 +56,10 @@ int mainEntryClickHouseClusterCopier(int argc, char ** argv);
 #if ENABLE_CLICKHOUSE_OBFUSCATOR
 int mainEntryClickHouseObfuscator(int argc, char ** argv);
 #endif
+#if ENABLE_CLICKHOUSE_ODBC_BRIDGE || !defined(ENABLE_CLICKHOUSE_ODBC_BRIDGE)
+int mainEntryClickHouseODBCBridge(int argc, char ** argv);
+#endif
+
 
 #if USE_EMBEDDED_COMPILER
     int mainEntryClickHouseClang(int argc, char ** argv);
@@ -101,6 +105,10 @@ std::pair<const char *, MainFunc> clickhouse_applications[] =
 #if ENABLE_CLICKHOUSE_OBFUSCATOR
     {"obfuscator", mainEntryClickHouseObfuscator},
 #endif
+#if ENABLE_CLICKHOUSE_ODBC_BRIDGE || !defined(ENABLE_CLICKHOUSE_ODBC_BRIDGE)
+    {"odbc-bridge", mainEntryClickHouseODBCBridge},
+#endif
+
 #if USE_EMBEDDED_COMPILER
     {"clang", mainEntryClickHouseClang},
     {"clang++", mainEntryClickHouseClang},

dbms/programs/odbc-bridge/CMakeLists.txt

@@ -0,0 +1,12 @@
+add_library (clickhouse-odbc-bridge-lib
+    ODBCBridge.cpp
+    ODBCHandler.cpp
+    )
+
+target_link_libraries (clickhouse-odbc-bridge-lib clickhouse_common_io daemon)
+target_include_directories (clickhouse-odbc-bridge-lib PUBLIC ${ClickHouse_SOURCE_DIR}/libs/libdaemon/include)
+
+if (CLICKHOUSE_SPLIT_BINARY)
+    add_executable (clickhouse-odbc-bridge odbc-bridge.cpp)
+    target_link_libraries (clickhouse-odbc-bridge clickhouse-odbc-bridge-lib)
+endif ()

dbms/programs/odbc-bridge/ODBCBridge.cpp

@@ -0,0 +1,193 @@
+#include "ODBCBridge.h"
+#include "ODBCHandler.h"
+
+#include <string>
+#include <errno.h>
+#include <IO/ReadHelpers.h>
+#include <boost/program_options.hpp>
+#include <Poco/Net/HTTPServer.h>
+#include <Poco/Net/NetException.h>
+#include <Poco/String.h>
+#include <Poco/Util/HelpFormatter.h>
+#include <Common/Exception.h>
+#include <Common/StringUtils/StringUtils.h>
+#include <Common/config.h>
+#include <common/logger_useful.h>
+
+namespace DB
+{
+namespace ErrorCodes
+{
+    extern const int ARGUMENT_OUT_OF_BOUND;
+}
+
+namespace
+{
+    std::string getCanonicalPath(std::string && path)
+    {
+        Poco::trimInPlace(path);
+        if (path.empty())
+            throw Exception("path configuration parameter is empty");
+        if (path.back() != '/')
+            path += '/';
+        return std::move(path);
+    }
+
+    Poco::Net::SocketAddress makeSocketAddress(const std::string & host, UInt16 port, Poco::Logger * log)
+    {
+        Poco::Net::SocketAddress socket_address;
+        try
+        {
+            socket_address = Poco::Net::SocketAddress(host, port);
+        }
+        catch (const Poco::Net::DNSException & e)
+        {
+            const auto code = e.code();
+            if (code == EAI_FAMILY
+#if defined(EAI_ADDRFAMILY)
+                || code == EAI_ADDRFAMILY
+#endif
+            )
+            {
+                LOG_ERROR(log,
+                    "Cannot resolve listen_host (" << host << "), error " << e.code() << ": " << e.message()
+                                                   << ". "
+                                                      "If it is an IPv6 address and your host has disabled IPv6, then consider to "
+                                                      "specify IPv4 address to listen in <listen_host> element of configuration "
+                                                      "file. Example: <listen_host>0.0.0.0</listen_host>");
+            }
+
+            throw;
+        }
+        return socket_address;
+    }
+
+    Poco::Net::SocketAddress socketBindListen(Poco::Net::ServerSocket & socket, const std::string & host, UInt16 port, Poco::Logger * log)
+    {
+        auto address = makeSocketAddress(host, port, log);
+#if POCO_VERSION < 0x01080000
+        socket.bind(address, /* reuseAddress = */ true);
+#else
+        socket.bind(address, /* reuseAddress = */ true, /* reusePort = */ false);
+#endif
+
+        socket.listen(/* backlog = */ 64);
+
+        return address;
+    };
+}
+
+std::string ODBCBridge::getDefaultCorePath() const
+{
+    return getCanonicalPath(config().getString("path")) + "cores";
+}
+
+void ODBCBridge::handleHelp(const std::string &, const std::string &)
+{
+    Poco::Util::HelpFormatter helpFormatter(options());
+    helpFormatter.setCommand(commandName());
+    helpFormatter.setHeader("HTTP-proxy for odbc requests");
+    helpFormatter.setUsage("--http-port <port>");
+    helpFormatter.format(std::cerr);
+
+    stopOptionsProcessing();
+}
+
+
+void ODBCBridge::defineOptions(Poco::Util::OptionSet & options)
+{
+    options.addOption(Poco::Util::Option("http-port", "", "port to listen").argument("http-port", true).binding("http-port"));
+    options.addOption(
+        Poco::Util::Option("http-host", "", "hostname to listen, default localhost").argument("http-host").binding("http-host"));
+    options.addOption(
+        Poco::Util::Option("http-timeout", "", "http timout for socket, default 1800").argument("http-timeout").binding("http-timeout"));
+    options.addOption(Poco::Util::Option("log-level", "", "sets log level, default info").argument("log-level").binding("log-level"));
+    options.addOption(Poco::Util::Option("max-server-connections", "", "max connections to server, default 1024")
+                          .argument("max-server-connections")
+                          .binding("max-server-connections"));
+    options.addOption(Poco::Util::Option("keep-alive-timeout", "", "keepalive timeout, default 10")
+                          .argument("keep-alive-timeout")
+                          .binding("keep-alive-timeout"));
+
+    using Me = std::decay_t<decltype(*this)>;
+    options.addOption(Poco::Util::Option("help", "", "produce this help message")
+                          .binding("help")
+                          .callback(Poco::Util::OptionCallback<Me>(this, &Me::handleHelp)));
+
+    ServerApplication::defineOptions(options); /// Don't need complex .xml config
+}
+
+void ODBCBridge::initialize(Application & self)
+{
+    is_help = config().has("help");
+    if (is_help)
+        return;
+
+    log_level = config().getString("log-level", "info");
+    Poco::Logger::root().setLevel(log_level);
+    log = &logger();
+    hostname = config().getString("http-host", "localhost");
+    port = config().getUInt("http-port");
+    if (port < 0 || port > 0xFFFF)
+        throw Exception("Out of range 'http-port': " + std::to_string(port), ErrorCodes::ARGUMENT_OUT_OF_BOUND);
+
+    http_timeout = config().getUInt("http-timeout", DEFAULT_HTTP_READ_BUFFER_TIMEOUT);
+    max_server_connections = config().getUInt("max-server-connections", 1024);
+    keep_alive_timeout = config().getUInt("keep-alive-timeout", 10);
+
+    initializeTerminationAndSignalProcessing();
+
+    ServerApplication::initialize(self);
+}
+
+void ODBCBridge::uninitialize()
+{
+    LOG_INFO(log, "Shutting down");
+    BaseDaemon::uninitialize();
+}
+
+int ODBCBridge::main(const std::vector<std::string> & /*args*/)
+{
+    if (is_help)
+        return 0;
+
+    LOG_INFO(log, "Starting up");
+    Poco::Net::ServerSocket socket;
+    auto address = socketBindListen(socket, hostname, port, log);
+    socket.setReceiveTimeout(http_timeout);
+    socket.setSendTimeout(http_timeout);
+    Poco::ThreadPool server_pool(3, max_server_connections);
+    Poco::Net::HTTPServerParams::Ptr http_params = new Poco::Net::HTTPServerParams;
+    http_params->setTimeout(http_timeout);
+    http_params->setKeepAliveTimeout(keep_alive_timeout);
+
+    context = std::make_shared<Context>(Context::createGlobal());
+    context->setGlobalContext(*context);
+
+    auto server = Poco::Net::HTTPServer(
+        new ODBCRequestHandlerFactory("ODBCRequestHandlerFactory-factory", keep_alive_timeout, context), server_pool, socket, http_params);
+    server.start();
+
+    LOG_INFO(log, "Listening http://" + address.toString());
+
+    waitForTerminationRequest();
+
+
+    return Application::EXIT_OK;
+}
+}
+
+int mainEntryClickHouseODBCBridge(int argc, char ** argv)
+{
+    DB::ODBCBridge app;
+    try
+    {
+        return app.run(argc, argv);
+    }
+    catch (...)
+    {
+        std::cerr << DB::getCurrentExceptionMessage(true) << "\n";
+        auto code = DB::getCurrentExceptionCode();
+        return code ? code : 1;
+    }
+}

dbms/programs/odbc-bridge/ODBCBridge.h

@@ -0,0 +1,41 @@
+#pragma once
+
+#include <daemon/BaseDaemon.h>
+#include <Poco/Logger.h>
+#include <Interpreters/Context.h>
+
+namespace DB
+{
+class ODBCBridge : public BaseDaemon
+{
+
+public:
+    void defineOptions(Poco::Util::OptionSet & options) override;
+
+protected:
+    void initialize(Application & self) override;
+
+    void uninitialize() override;
+
+    int main(const std::vector<std::string> & args) override;
+
+    std::string getDefaultCorePath() const override;
+
+private:
+
+    void handleHelp(const std::string &, const std::string &);
+
+    bool is_help;
+    std::string hostname;
+    UInt16 port;
+    size_t http_timeout;
+    std::string log_level;
+    size_t max_server_connections;
+    size_t keep_alive_timeout;
+
+    Poco::Logger * log;
+
+    std::shared_ptr<Context> context; /// need for settings only
+};
+
+}

dbms/programs/odbc-bridge/ODBCHandler.cpp

@@ -0,0 +1,137 @@
+#include "ODBCHandler.h"
+#include <Common/HTMLForm.h>
+
+#include <memory>
+#include <DataStreams/IBlockOutputStream.h>
+#include <DataTypes/DataTypeFactory.h>
+#include <Dictionaries/ODBCBlockInputStream.h>
+#include <Dictionaries/validateODBCConnectionString.h>
+#include <Formats/BinaryRowInputStream.h>
+#include <Formats/FormatFactory.h>
+#include <IO/ReadBufferFromIStream.h>
+#include <IO/WriteBufferFromHTTPServerResponse.h>
+#include <IO/WriteHelpers.h>
+#include <Interpreters/Context.h>
+#include <boost/algorithm/string.hpp>
+#include <boost/tokenizer.hpp>
+#include <Poco/Ext/SessionPoolHelpers.h>
+#include <Poco/Net/HTTPBasicCredentials.h>
+#include <Poco/Net/HTTPServerRequest.h>
+#include <Poco/Net/HTTPServerResponse.h>
+#include <common/logger_useful.h>
+namespace DB
+{
+namespace ErrorCodes
+{
+    extern const int BAD_REQUEST_PARAMETER;
+}
+
+namespace
+{
+    std::string buildConnectionString(const std::string & DSN, const std::string & database)
+    {
+        std::stringstream ss;
+        ss << "DSN=" << DSN << ";DATABASE=" << database;
+        return ss.str();
+    }
+    NamesAndTypesList parseColumns(std::string && column_string)
+    {
+        const auto & factory_instance = DataTypeFactory::instance();
+        NamesAndTypesList result;
+        static boost::char_separator<char> sep(",");
+        boost::tokenizer<boost::char_separator<char>> tokens(column_string, sep);
+        for (const std::string & name_and_type_str : tokens)
+        {
+            std::vector<std::string> name_and_type;
+            boost::split(name_and_type, name_and_type_str, boost::is_any_of(":"));
+            if (name_and_type.size() != 2)
+                throw Exception("ODBCBridge: Invalid columns parameter '" + column_string + "'", ErrorCodes::BAD_REQUEST_PARAMETER);
+            result.emplace_back(name_and_type[0], factory_instance.get(name_and_type[1]));
+        }
+        return result;
+    }
+}
+void ODBCHTTPHandler::handleRequest(Poco::Net::HTTPServerRequest & request, Poco::Net::HTTPServerResponse & response)
+{
+    Poco::Net::HTMLForm params(request, request.stream());
+    LOG_TRACE(log, "Request URI: " + request.getURI());
+
+    if (!params.has("query"))
+        throw Exception("ODBCBridge: No 'query' in request body", ErrorCodes::BAD_REQUEST_PARAMETER);
+
+    std::string query = params.get("query");
+
+    if (!params.has("columns"))
+        throw Exception("ODBCBridge: No 'columns' in request body", ErrorCodes::BAD_REQUEST_PARAMETER);
+
+    std::string columns = params.get("columns");
+
+    auto names_and_types = parseColumns(std::move(columns));
+    Block sample_block;
+    for (const NameAndTypePair & column_data : names_and_types)
+    {
+        sample_block.insert({column_data.type, column_data.name});
+    }
+
+    ODBCBlockInputStream inp(pool->get(), query, sample_block, max_block_size);
+
+    WriteBufferFromHTTPServerResponse out(request, response, keep_alive_timeout);
+    std::shared_ptr<IBlockOutputStream> writer = FormatFactory::instance().getOutput(format, out, sample_block, *context);
+
+    writer->writePrefix();
+    while (auto block = inp.read())
+        writer->write(block);
+
+    writer->writeSuffix();
+
+
+    writer->flush();
+}
+
+
+void PingHandler::handleRequest(Poco::Net::HTTPServerRequest & /*request*/, Poco::Net::HTTPServerResponse & response)
+{
+    try
+    {
+        setResponseDefaultHeaders(response, keep_alive_timeout);
+        const char * data = "Ok.\n";
+        response.sendBuffer(data, strlen(data));
+    }
+    catch (...)
+    {
+        tryLogCurrentException("PingHandler");
+    }
+}
+
+Poco::Net::HTTPRequestHandler * ODBCRequestHandlerFactory::createRequestHandler(const Poco::Net::HTTPServerRequest & request)
+{
+    const auto & uri = request.getURI();
+    LOG_TRACE(log, "Request URI: " + uri);
+
+    if (uri == "/ping" && request.getMethod() == Poco::Net::HTTPRequest::HTTP_GET)
+        return new PingHandler(keep_alive_timeout);
+
+    HTMLForm params(request);
+    std::string DSN = params.get("DSN");
+    std::string database = params.get("database");
+    std::string max_block_size = params.get("max_block_size", "");
+    std::string format = params.get("format", "RowBinary");
+
+    std::string connection_string = buildConnectionString(DSN, database);
+
+    LOG_TRACE(log, "Connection string:" << connection_string);
+
+    if (!pool_map.count(connection_string))
+        pool_map[connection_string] = createAndCheckResizePocoSessionPool(
+            [&] { return std::make_shared<Poco::Data::SessionPool>("ODBC", validateODBCConnectionString(connection_string)); });
+
+    if (request.getMethod() == Poco::Net::HTTPRequest::HTTP_POST)
+        return new ODBCHTTPHandler(pool_map.at(connection_string),
+            format,
+            max_block_size == "" ? DEFAULT_BLOCK_SIZE : parse<size_t>(max_block_size),
+            keep_alive_timeout,
+            context);
+
+    return nullptr;
+}
+}

dbms/programs/odbc-bridge/ODBCHandler.h

@@ -0,0 +1,70 @@
+#pragma once
+
+#include <Interpreters/Context.h>
+#include <Poco/Logger.h>
+#include <Poco/Net/HTTPRequestHandler.h>
+#include <Poco/Net/HTTPRequestHandlerFactory.h>
+
+#pragma GCC diagnostic push
+#pragma GCC diagnostic ignored "-Wunused-parameter"
+#include <Poco/Data/SessionPool.h>
+#pragma GCC diagnostic pop
+
+namespace DB
+{
+class ODBCHTTPHandler : public Poco::Net::HTTPRequestHandler
+{
+public:
+    ODBCHTTPHandler(std::shared_ptr<Poco::Data::SessionPool> pool_,
+        const std::string & format_,
+        size_t max_block_size_,
+        size_t keep_alive_timeout_,
+        std::shared_ptr<Context> context_)
+        : log(&Poco::Logger::get("ODBCHTTPHandler"))
+        , pool(pool_)
+        , format(format_)
+        , max_block_size(max_block_size_)
+        , keep_alive_timeout(keep_alive_timeout_)
+        , context(context_)
+    {
+    }
+
+    void handleRequest(Poco::Net::HTTPServerRequest & request, Poco::Net::HTTPServerResponse & response) override;
+
+private:
+    Poco::Logger * log;
+    std::shared_ptr<Poco::Data::SessionPool> pool;
+    std::string format;
+    size_t max_block_size;
+    size_t keep_alive_timeout;
+    std::shared_ptr<Context> context;
+};
+
+class PingHandler : public Poco::Net::HTTPRequestHandler
+{
+public:
+    PingHandler(size_t keep_alive_timeout_) : keep_alive_timeout(keep_alive_timeout_) {}
+    void handleRequest(Poco::Net::HTTPServerRequest & request, Poco::Net::HTTPServerResponse & response) override;
+
+private:
+    size_t keep_alive_timeout;
+};
+
+class ODBCRequestHandlerFactory : public Poco::Net::HTTPRequestHandlerFactory
+{
+public:
+    ODBCRequestHandlerFactory(const std::string & name_, size_t keep_alive_timeout_, std::shared_ptr<Context> context_)
+        : log(&Poco::Logger::get(name_)), name(name_), keep_alive_timeout(keep_alive_timeout_), context(context_)
+    {
+    }
+
+    Poco::Net::HTTPRequestHandler * createRequestHandler(const Poco::Net::HTTPServerRequest & request) override;
+
+private:
+    Poco::Logger * log;
+    std::string name;
+    size_t keep_alive_timeout;
+    std::shared_ptr<Context> context;
+    std::unordered_map<std::string, std::shared_ptr<Poco::Data::SessionPool>> pool_map;
+};
+}

dbms/programs/odbc-bridge/odbc-bridge.cpp

@@ -0,0 +1,2 @@
+int mainEntryClickHouseODBCBridge(int argc, char ** argv);
+int main(int argc_, char ** argv_) { return mainEntryOdbcBridge(argc_, argv_); }

dbms/src/Common/ErrorCodes.cpp

@@ -379,6 +379,7 @@ namespace ErrorCodes
     extern const int CANNOT_IOSETUP = 402;
     extern const int INVALID_JOIN_ON_EXPRESSION = 403;
     extern const int BAD_ODBC_CONNECTION_STRING = 404;
+    extern const int BAD_REQUEST_PARAMETER = 405;
 
     extern const int KEEPER_EXCEPTION = 999;
     extern const int POCO_EXCEPTION = 1000;

libs/libdaemon/include/daemon/BaseDaemon.h

@@ -159,6 +159,9 @@ protected:
     /// thread safe
     virtual void handleSignal(int signal_id);
 
+    /// initialize termination process and signal handlers
+    virtual void initializeTerminationAndSignalProcessing();
+
     /// реализация обработки сигналов завершения через pipe не требует блокировки сигнала с помощью sigprocmask во всех потоках
     void waitForTerminationRequest()
 #if POCO_CLICKHOUSE_PATCH || POCO_VERSION >= 0x02000000 // in old upstream poco not vitrual

libs/libdaemon/src/BaseDaemon.cpp

@@ -1031,6 +1031,19 @@ void BaseDaemon::initialize(Application & self)
             throw Poco::Exception("Cannot change directory to " + core_path);
     }
 
+    initializeTerminationAndSignalProcessing();
+
+    logRevision();
+
+    for (const auto & key : DB::getMultipleKeysFromConfig(config(), "", "graphite"))
+    {
+        graphite_writers.emplace(key, std::make_unique<GraphiteWriter>(key));
+    }
+}
+
+
+void BaseDaemon::initializeTerminationAndSignalProcessing()
+{
     std::set_terminate(terminate_handler);
 
     /// We want to avoid SIGPIPE when working with sockets and pipes, and just handle return value/errno instead.
@@ -1071,15 +1084,9 @@ void BaseDaemon::initialize(Application & self)
     static KillingErrorHandler killing_error_handler;
     Poco::ErrorHandler::set(&killing_error_handler);
 
-    logRevision();
-
     signal_listener.reset(new SignalListener(*this));
     signal_listener_thread.start(*signal_listener);
 
-    for (const auto & key : DB::getMultipleKeysFromConfig(config(), "", "graphite"))
-    {
-        graphite_writers.emplace(key, std::make_unique<GraphiteWriter>(key));
-    }
 }
 
 void BaseDaemon::logRevision() const