CLICKHOUSE-3878: Next iteration in odbc-bridge

dbms/src/Common/ErrorCodes.cpp

@@ -381,6 +381,7 @@ namespace ErrorCodes
     extern const int BAD_ODBC_CONNECTION_STRING = 404;
     extern const int PARTITION_SIZE_EXCEEDS_MAX_DROP_SIZE_LIMIT = 405;
     extern const int BAD_REQUEST_PARAMETER = 406;
+    extern const int EXTERNAL_EXECUTABLE_NOT_FOUND = 407;
 
     extern const int KEEPER_EXCEPTION = 999;
     extern const int POCO_EXCEPTION = 1000;

dbms/src/Storages/StorageODBC.cpp

-#include <Poco/Ext/SessionPoolHelpers.h>
-#include <Storages/transformQueryForExternalDatabase.h>
-#include <Storages/StorageODBC.h>
-#include <Storages/StorageFactory.h>
-#include <Interpreters/evaluateConstantExpression.h>
 #include <Dictionaries/ODBCBlockInputStream.h>
 #include <Dictionaries/validateODBCConnectionString.h>
+#include <Interpreters/Context.h>
+#include <Interpreters/evaluateConstantExpression.h>
 #include <Parsers/ASTLiteral.h>
+#include <Storages/StorageFactory.h>
+#include <Storages/StorageODBC.h>
+#include <Storages/transformQueryForExternalDatabase.h>
+#include <Poco/Ext/SessionPoolHelpers.h>
+#include <Poco/Util/AbstractConfiguration.h>
+#include <common/logger_useful.h>
 
-
+#include <IO/ReadHelpers.h>
+#include <IO/ReadWriteBufferFromHTTP.h>
+#include <Poco/File.h>
+#include <Poco/Net/HTTPRequest.h>
+#include <Poco/Path.h>
+#include <Common/ShellCommand.h>
 namespace DB
 {
-
 namespace ErrorCodes
 {
     extern const int NUMBER_OF_ARGUMENTS_DOESNT_MATCH;
+    extern const int EXTERNAL_EXECUTABLE_NOT_FOUND;
 }
 
 
-StorageODBC::StorageODBC(
-    const std::string & name,
+StorageODBC::StorageODBC(const std::string & table_name_,
     const std::string & connection_string,
-    const std::string & remote_database_name,
-    const std::string & remote_table_name,
-    const ColumnsDescription & columns_)
-    : IStorage{columns_}
-    , name(name)
-    , remote_database_name(remote_database_name)
-    , remote_table_name(remote_table_name)
+    const std::string & remote_database_name_,
+    const std::string & remote_table_name_,
+    const ColumnsDescription & columns_,
+    const Context & context_)
+    : IStorageURLBase(Poco::URI(), context_, table_name_, "RowBinary", columns_)
+    , connection_string(connection_string)
+    , remote_database_name(remote_database_name_)
+    , remote_table_name(remote_table_name_)
+{
+    const auto & config = context_.getConfigRef();
+    size_t bridge_port = config.getUInt("odbc_bridge.port", 9018);
+    std::string bridge_host = config.getString("odbc_bridge.host", "localhost");
+
+    uri.setHost(bridge_host);
+    uri.setPort(bridge_port);
+    uri.setScheme("http");
+
+    ping_uri = uri;
+    ping_uri.setPath("/ping");
+}
+
+std::string StorageODBC::getReadMethod() const
+{
+    return Poco::Net::HTTPRequest::HTTP_POST;
+}
+
+std::vector<std::pair<std::string, std::string>> StorageODBC::getReadURIParams(const Names & column_names,
+    const SelectQueryInfo & /*query_info*/,
+    const Context & /*context*/,
+    QueryProcessingStage::Enum & /*processed_stage*/,
+    size_t max_block_size) const
 {
-    pool = createAndCheckResizePocoSessionPool([&]
+    NamesAndTypesList cols;
+    for (const String & name : column_names)
     {
-        return std::make_shared<Poco::Data::SessionPool>("ODBC", validateODBCConnectionString(connection_string));
-    });
+        auto column_data = getColumn(name);
+        cols.emplace_back(column_data.name, column_data.type);
+    }
+    std::vector<std::pair<std::string, std::string>> result;
+
+    result.emplace_back("connection_string", connection_string);
+    result.emplace_back("columns", cols.toString());
+    result.emplace_back("max_block_size", std::to_string(max_block_size));
+
+    return result;
 }
 
-BlockInputStreams StorageODBC::read(
-    const Names & column_names,
+std::function<void(std::ostream &)> StorageODBC::getReadPOSTDataCallback(const Names & /*column_names*/,
     const SelectQueryInfo & query_info,
     const Context & context,
-    QueryProcessingStage::Enum & processed_stage,
-    size_t max_block_size,
-    unsigned /*num_streams*/)
+    QueryProcessingStage::Enum & /*processed_stage*/,
+    size_t /*max_block_size*/) const
 {
-    check(column_names);
-    processed_stage = QueryProcessingStage::FetchColumns;
     String query = transformQueryForExternalDatabase(
         *query_info.query, getColumns().ordinary, IdentifierQuotingStyle::DoubleQuotes, remote_database_name, remote_table_name, context);
 
-    Block sample_block;
-    for (const String & name : column_names)
+    return [query](std::ostream & os) { os << "query=" << query; };
+}
+
+bool StorageODBC::checkODBCBridgeIsRunning() const
+{
+    try
     {
-        auto column_data = getColumn(name);
-        sample_block.insert({ column_data.type, column_data.name });
+        ReadWriteBufferFromHTTP buf(ping_uri, Poco::Net::HTTPRequest::HTTP_GET, nullptr);
+        return checkString("Ok.", buf);
+    }
+    catch (...)
+    {
+        return false;
+    }
+}
+
+
+void StorageODBC::startODBCBridge() const
+{
+    const auto & config = context_global.getConfigRef();
+    const auto & settings = context_global.getSettingsRef();
+    Poco::Path path{config.getString("application.dir", "")};
+    path.setFileName("clickhouse-odbc-bridge");
+
+    if (!path.isFile())
+        throw Exception("clickhouse-odbc-bridge is not found", ErrorCodes::EXTERNAL_EXECUTABLE_NOT_FOUND);
+
+    std::stringstream command;
+    command << path.toString() << ' ';
+    command << "--daemon" << ' ';
+    command << "--http-port " << config.getUInt("odbc_bridge.port", 9018) << ' ';
+    command << "--http-host " << config.getString("odbc_bridge.host", "localhost") << ' ';
+    command << "--http-timeout " << settings.http_receive_timeout.value.totalSeconds() << ' ';
+    if (config.has("logger.odbc_bridge_log"))
+        command << "--log-path " << config.getString("logger.odbc_bridge_log") << ' ';
+    if (config.has("logger.odbc_bridge_errlog"))
+        command << "--err-log-path " << config.getString("logger.odbc_bridge_errlog") << ' ';
+    if (config.has("logger.odbc_bridge_level"))
+        command << "--log-level " << config.getString("logger.odbc_bridge_level") << ' ';
+
+    auto cmd = ShellCommand::execute(command.str());
+    cmd->wait();
+}
+BlockInputStreams StorageODBC::read(const Names & column_names,
+    const SelectQueryInfo & query_info,
+    const Context & context,
+    QueryProcessingStage::Enum & processed_stage,
+    size_t max_block_size,
+    unsigned num_streams)
+{
+    if (!checkODBCBridgeIsRunning())
+    {
+        startODBCBridge();
+        size_t counter = 0;
+        while (!checkODBCBridgeIsRunning() && counter <= 5)
+        {
+            sleep(1);
+            counter++;
+        }
     }
 
-    return { std::make_shared<ODBCBlockInputStream>(pool->get(), query, sample_block, max_block_size) };
+    return IStorageURLBase::read(column_names, query_info, context, processed_stage, max_block_size, num_streams);
 }
 
 
 void registerStorageODBC(StorageFactory & factory)
 {
-    factory.registerStorage("ODBC", [](const StorageFactory::Arguments & args)
-    {
+    factory.registerStorage("ODBC", [](const StorageFactory::Arguments & args) {
         ASTs & engine_args = args.engine_args;
 
         if (engine_args.size() != 3)
             throw Exception(
-                "Storage ODBC requires exactly 3 parameters: ODBC('DSN', database, table).",
-                ErrorCodes::NUMBER_OF_ARGUMENTS_DOESNT_MATCH);
+                "Storage ODBC requires exactly 3 parameters: ODBC('DSN', database, table).", ErrorCodes::NUMBER_OF_ARGUMENTS_DOESNT_MATCH);
 
         for (size_t i = 0; i < 2; ++i)
             engine_args[i] = evaluateConstantExpressionOrIdentifierAsLiteral(engine_args[i], args.local_context);
@@ -76,8 +164,8 @@ void registerStorageODBC(StorageFactory & factory)
             static_cast<const ASTLiteral &>(*engine_args[0]).value.safeGet<String>(),
             static_cast<const ASTLiteral &>(*engine_args[1]).value.safeGet<String>(),
             static_cast<const ASTLiteral &>(*engine_args[2]).value.safeGet<String>(),
-            args.columns);
+            args.columns,
+            args.context);
     });
 }
-
 }

dbms/src/Storages/StorageODBC.h

 #pragma once
 
+#include <Storages/StorageURL.h>
 #include <ext/shared_ptr_helper.h>
-#include <Storages/IStorage.h>
-
-#pragma GCC diagnostic push
-#pragma GCC diagnostic ignored "-Wunused-parameter"
-    #include <Poco/Data/SessionPool.h>
-#pragma GCC diagnostic pop
-
-
 
 namespace DB
 {
-
 /** Implements storage in the ODBC database.
   * Use ENGINE = odbc(connection_string, table_name)
   * Example ENGINE = odbc('dsn=test', table)
   * Read only.
   */
-class StorageODBC : public ext::shared_ptr_helper<StorageODBC>, public IStorage
+class StorageODBC : public ext::shared_ptr_helper<StorageODBC>, public IStorageURLBase
 {
 public:
-    StorageODBC(
-        const std::string & name,
-        const std::string & connection_string,
-        const std::string & remote_database_name,
-        const std::string & remote_table_name,
-        const ColumnsDescription & columns_);
-
-    std::string getName() const override { return "ODBC"; }
-    std::string getTableName() const override { return name; }
+    std::string getName() const override
+    {
+        return "ODBC";
+    }
 
-    BlockInputStreams read(
-        const Names & column_names,
+    BlockInputStreams read(const Names & column_names,
         const SelectQueryInfo & query_info,
         const Context & context,
         QueryProcessingStage::Enum & processed_stage,
         size_t max_block_size,
         unsigned num_streams) override;
 
+
+protected:
+    StorageODBC(const std::string & table_name_,
+        const std::string & connection_string,
+        const std::string & remote_database_name,
+        const std::string & remote_table_name,
+        const ColumnsDescription & columns_,
+        const Context & context_);
+
 private:
-    std::string name;
+    std::string connection_string;
     std::string remote_database_name;
     std::string remote_table_name;
+    Poco::URI ping_uri;
+
+    std::string getReadMethod() const override;
+
+    std::vector<std::pair<std::string, std::string>> getReadURIParams(const Names & column_names,
+        const SelectQueryInfo & query_info,
+        const Context & context,
+        QueryProcessingStage::Enum & processed_stage,
+        size_t max_block_size) const override;
+
+    std::function<void(std::ostream &)> getReadPOSTDataCallback(const Names & column_names,
+        const SelectQueryInfo & query_info,
+        const Context & context,
+        QueryProcessingStage::Enum & processed_stage,
+        size_t max_block_size) const override;
 
-    std::shared_ptr<Poco::Data::SessionPool> pool;
+    bool checkODBCBridgeIsRunning() const;
+    void startODBCBridge() const;
 };
 }

dbms/src/Storages/StorageURL.cpp

@@ -24,12 +24,12 @@ namespace ErrorCodes
     extern const int NUMBER_OF_ARGUMENTS_DOESNT_MATCH;
 };
 
-StorageURL::StorageURL(const Poco::URI & uri_,
+IStorageURLBase::IStorageURLBase(const Poco::URI & uri_,
+    const Context & context_,
     const std::string & table_name_,
     const String & format_name_,
-    const ColumnsDescription & columns_,
-    Context & context_)
-    : IStorage(columns_), uri(uri_), format_name(format_name_), table_name(table_name_), context_global(context_)
+    const ColumnsDescription & columns_)
+    : IStorage(columns_), uri(uri_), context_global(context_), format_name(format_name_), table_name(table_name_)
 {
 }
 
@@ -39,6 +39,8 @@ namespace
     {
     public:
         StorageURLBlockInputStream(const Poco::URI & uri,
+            const std::string & method,
+            std::function<void(std::ostream &)> callback,
             const String & format,
             const String & name_,
             const Block & sample_block,
@@ -47,7 +49,7 @@ namespace
             const ConnectionTimeouts & timeouts)
             : name(name_)
         {
-            read_buf = std::make_unique<ReadWriteBufferFromHTTP>(uri, Poco::Net::HTTPRequest::HTTP_GET, nullptr, timeouts);
+            read_buf = std::make_unique<ReadWriteBufferFromHTTP>(uri, method, callback, timeouts);
 
             reader = FormatFactory::instance().getInput(format, *read_buf, sample_block, context, max_block_size);
         }
@@ -89,7 +91,7 @@ namespace
         StorageURLBlockOutputStream(const Poco::URI & uri,
             const String & format,
             const Block & sample_block_,
-            Context & context,
+            const Context & context,
             const ConnectionTimeouts & timeouts)
             : sample_block(sample_block_)
         {
@@ -127,16 +129,45 @@ namespace
 }
 
 
-BlockInputStreams StorageURL::read(
-    const Names & /*column_names*/,
+std::string IStorageURLBase::getReadMethod() const
+{
+    return Poco::Net::HTTPRequest::HTTP_GET;
+}
+
+std::vector<std::pair<std::string, std::string>> IStorageURLBase::getReadURIParams(const Names & /*column_names*/,
     const SelectQueryInfo & /*query_info*/,
-    const Context & context,
+    const Context & /*context*/,
     QueryProcessingStage::Enum & /*processed_stage*/,
+    size_t /*max_block_size*/) const
+{
+    return {};
+}
+
+std::function<void(std::ostream &)> IStorageURLBase::getReadPOSTDataCallback(const Names & /*column_names*/,
+    const SelectQueryInfo & /*query_info*/,
+    const Context & /*context*/,
+    QueryProcessingStage::Enum & /*processed_stage*/,
+    size_t /*max_block_size*/) const
+{
+    return nullptr;
+}
+
+
+BlockInputStreams IStorageURLBase::read(const Names & column_names,
+    const SelectQueryInfo & query_info,
+    const Context & context,
+    QueryProcessingStage::Enum & processed_stage,
     size_t max_block_size,
     unsigned /*num_streams*/)
 {
-    return {std::make_shared<StorageURLBlockInputStream>(
-        uri,
+    auto request_uri = uri;
+    auto params = getReadURIParams(column_names, query_info, context, processed_stage, max_block_size);
+    for (const auto & [param, value] : params)
+        request_uri.addQueryParameter(param, value);
+
+    return {std::make_shared<StorageURLBlockInputStream>(request_uri,
+        getReadMethod(),
+        getReadPOSTDataCallback(column_names, query_info, context, processed_stage, max_block_size),
         format_name,
         getName(),
         getSampleBlock(),
@@ -145,9 +176,9 @@ BlockInputStreams StorageURL::read(
         ConnectionTimeouts::getHTTPTimeouts(context.getSettingsRef()))};
 }
 
-void StorageURL::rename(const String & /*new_path_to_db*/, const String & /*new_database_name*/, const String & /*new_table_name*/) {}
+void IStorageURLBase::rename(const String & /*new_path_to_db*/, const String & /*new_database_name*/, const String & /*new_table_name*/) {}
 
-BlockOutputStreamPtr StorageURL::write(const ASTPtr & /*query*/, const Settings & /*settings*/)
+BlockOutputStreamPtr IStorageURLBase::write(const ASTPtr & /*query*/, const Settings & /*settings*/)
 {
     return std::make_shared<StorageURLBlockOutputStream>(
         uri, format_name, getSampleBlock(), context_global, ConnectionTimeouts::getHTTPTimeouts(context_global.getSettingsRef()));
@@ -155,8 +186,7 @@ BlockOutputStreamPtr StorageURL::write(const ASTPtr & /*query*/, const Settings
 
 void registerStorageURL(StorageFactory & factory)
 {
-    factory.registerStorage("URL", [](const StorageFactory::Arguments & args)
-    {
+    factory.registerStorage("URL", [](const StorageFactory::Arguments & args) {
         ASTs & engine_args = args.engine_args;
 
         if (!(engine_args.size() == 1 || engine_args.size() == 2))
@@ -175,5 +205,4 @@ void registerStorageURL(StorageFactory & factory)
         return StorageURL::create(uri, args.table_name, format_name, args.columns, args.context);
     });
 }
-
 }

dbms/src/Storages/StorageURL.h

@@ -13,14 +13,9 @@ namespace DB
  * HTTP POST when insert is called. In POST request the data is send
  * using Chunked transfer encoding, so server have to support it.
  */
-class StorageURL : public ext::shared_ptr_helper<StorageURL>, public IStorage
+class IStorageURLBase : public IStorage
 {
 public:
-    String getName() const override
-    {
-        return "URL";
-    }
-
     String getTableName() const override
     {
         return table_name;
@@ -38,18 +33,49 @@ public:
     void rename(const String & new_path_to_db, const String & new_database_name, const String & new_table_name) override;
 
 protected:
-    StorageURL(const Poco::URI & uri_,
+    IStorageURLBase(const Poco::URI & uri_,
+        const Context & context_,
         const std::string & table_name_,
         const String & format_name_,
-        const ColumnsDescription & columns_,
-        Context & context_);
+        const ColumnsDescription & columns_);
 
-private:
     Poco::URI uri;
+    const Context & context_global;
+
+private:
     String format_name;
     String table_name;
-    Context & context_global;
 
-    Logger * log = &Logger::get("StorageURL");
+    virtual std::string getReadMethod() const;
+
+    virtual std::vector<std::pair<std::string, std::string>> getReadURIParams(const Names & column_names,
+        const SelectQueryInfo & query_info,
+        const Context & context,
+        QueryProcessingStage::Enum & processed_stage,
+        size_t max_block_size) const;
+
+    virtual std::function<void(std::ostream &)> getReadPOSTDataCallback(const Names & column_names,
+        const SelectQueryInfo & query_info,
+        const Context & context,
+        QueryProcessingStage::Enum & processed_stage,
+        size_t max_block_size) const;
+};
+
+class StorageURL : public ext::shared_ptr_helper<StorageURL>, public IStorageURLBase
+{
+public:
+    StorageURL(const Poco::URI & uri_,
+        const std::string & table_name_,
+        const String & format_name_,
+        const ColumnsDescription & columns_,
+        Context & context_)
+        : IStorageURLBase(uri_, context_, table_name_, format_name_, columns_)
+    {
+    }
+
+    String getName() const override
+    {
+        return "URL";
+    }
 };
 }

dbms/src/TableFunctions/TableFunctionODBC.cpp

@@ -114,7 +114,7 @@ StoragePtr TableFunctionODBC::executeImpl(const ASTPtr & ast_function, const Con
         columns.emplace_back(reinterpret_cast<char *>(column_name), getDataType(type));
     }
 
-    auto result = StorageODBC::create(table_name, connection_string, "", table_name, ColumnsDescription{columns});
+    auto result = StorageODBC::create(table_name, connection_string, "", table_name, ColumnsDescription{columns}, context);
     result->startup();
     return result;
 }