2018-08-19 17:09:54 +00:00
|
|
|
#include "MainHandler.h"
|
|
|
|
|
|
|
|
#include "validateODBCConnectionString.h"
|
2018-08-08 16:15:29 +00:00
|
|
|
#include <memory>
|
2018-08-09 12:57:34 +00:00
|
|
|
#include <DataStreams/copyData.h>
|
2018-08-08 16:15:29 +00:00
|
|
|
#include <DataTypes/DataTypeFactory.h>
|
2019-02-11 21:43:29 +00:00
|
|
|
#include "ODBCBlockInputStream.h"
|
2018-08-08 16:15:29 +00:00
|
|
|
#include <Formats/FormatFactory.h>
|
|
|
|
#include <IO/WriteBufferFromHTTPServerResponse.h>
|
|
|
|
#include <IO/WriteHelpers.h>
|
2018-09-14 19:48:51 +00:00
|
|
|
#include <IO/ReadHelpers.h>
|
2018-08-08 16:15:29 +00:00
|
|
|
#include <Interpreters/Context.h>
|
|
|
|
#include <Poco/Net/HTTPServerRequest.h>
|
|
|
|
#include <Poco/Net/HTTPServerResponse.h>
|
2018-09-14 19:48:51 +00:00
|
|
|
#include <Poco/Net/HTMLForm.h>
|
2018-08-08 16:15:29 +00:00
|
|
|
#include <common/logger_useful.h>
|
2019-04-17 17:36:58 +00:00
|
|
|
#include <mutex>
|
|
|
|
#include <Poco/ThreadPool.h>
|
2018-08-19 17:09:54 +00:00
|
|
|
|
2018-08-08 16:15:29 +00:00
|
|
|
namespace DB
|
|
|
|
{
|
|
|
|
namespace
|
|
|
|
{
|
2018-08-09 12:57:34 +00:00
|
|
|
std::unique_ptr<Block> parseColumns(std::string && column_string)
|
2018-08-08 16:15:29 +00:00
|
|
|
{
|
2018-08-09 12:57:34 +00:00
|
|
|
std::unique_ptr<Block> sample_block = std::make_unique<Block>();
|
|
|
|
auto names_and_types = NamesAndTypesList::parse(column_string);
|
|
|
|
for (const NameAndTypePair & column_data : names_and_types)
|
|
|
|
sample_block->insert({column_data.type, column_data.name});
|
|
|
|
return sample_block;
|
|
|
|
}
|
2018-08-08 16:15:29 +00:00
|
|
|
}
|
|
|
|
|
2019-04-17 17:36:58 +00:00
|
|
|
using PocoSessionPoolConstructor = std::function<std::shared_ptr<Poco::Data::SessionPool>()>;
|
|
|
|
/** Is used to adjust max size of default Poco thread pool. See issue #750
|
|
|
|
* Acquire the lock, resize pool and construct new Session.
|
|
|
|
*/
|
2019-12-15 06:34:43 +00:00
|
|
|
static std::shared_ptr<Poco::Data::SessionPool> createAndCheckResizePocoSessionPool(PocoSessionPoolConstructor pool_constr)
|
2019-04-17 17:36:58 +00:00
|
|
|
{
|
|
|
|
static std::mutex mutex;
|
|
|
|
|
|
|
|
Poco::ThreadPool & pool = Poco::ThreadPool::defaultPool();
|
|
|
|
|
|
|
|
/// NOTE: The lock don't guarantee that external users of the pool don't change its capacity
|
|
|
|
std::unique_lock lock(mutex);
|
|
|
|
|
|
|
|
if (pool.available() == 0)
|
|
|
|
pool.addCapacity(2 * std::max(pool.capacity(), 1));
|
|
|
|
|
|
|
|
return pool_constr();
|
|
|
|
}
|
2018-08-09 12:57:34 +00:00
|
|
|
|
|
|
|
ODBCHandler::PoolPtr ODBCHandler::getPool(const std::string & connection_str)
|
|
|
|
{
|
2018-08-10 11:42:12 +00:00
|
|
|
std::lock_guard lock(mutex);
|
2018-08-09 12:57:34 +00:00
|
|
|
if (!pool_map->count(connection_str))
|
|
|
|
{
|
2019-02-22 17:16:11 +00:00
|
|
|
pool_map->emplace(connection_str, createAndCheckResizePocoSessionPool([connection_str]
|
|
|
|
{
|
2018-08-19 17:09:54 +00:00
|
|
|
return std::make_shared<Poco::Data::SessionPool>("ODBC", validateODBCConnectionString(connection_str));
|
2018-08-09 12:57:34 +00:00
|
|
|
}));
|
|
|
|
}
|
|
|
|
return pool_map->at(connection_str);
|
|
|
|
}
|
|
|
|
|
2018-08-08 16:15:29 +00:00
|
|
|
void ODBCHandler::handleRequest(Poco::Net::HTTPServerRequest & request, Poco::Net::HTTPServerResponse & response)
|
|
|
|
{
|
|
|
|
Poco::Net::HTMLForm params(request, request.stream());
|
|
|
|
LOG_TRACE(log, "Request URI: " + request.getURI());
|
|
|
|
|
2018-08-24 00:07:25 +00:00
|
|
|
auto process_error = [&response, this](const std::string & message)
|
|
|
|
{
|
2018-08-08 16:15:29 +00:00
|
|
|
response.setStatusAndReason(Poco::Net::HTTPResponse::HTTP_INTERNAL_SERVER_ERROR);
|
|
|
|
if (!response.sent())
|
|
|
|
response.send() << message << std::endl;
|
|
|
|
LOG_WARNING(log, message);
|
|
|
|
};
|
|
|
|
|
|
|
|
if (!params.has("query"))
|
|
|
|
{
|
2018-08-13 15:00:41 +00:00
|
|
|
process_error("No 'query' in request body");
|
2018-08-08 16:15:29 +00:00
|
|
|
return;
|
|
|
|
}
|
|
|
|
|
|
|
|
if (!params.has("columns"))
|
|
|
|
{
|
2018-08-13 15:00:41 +00:00
|
|
|
process_error("No 'columns' in request URL");
|
2018-08-08 16:15:29 +00:00
|
|
|
return;
|
|
|
|
}
|
|
|
|
|
2018-08-09 12:57:34 +00:00
|
|
|
if (!params.has("connection_string"))
|
2018-08-08 16:15:29 +00:00
|
|
|
{
|
2018-08-13 15:00:41 +00:00
|
|
|
process_error("No 'connection_string' in request URL");
|
2018-08-08 16:15:29 +00:00
|
|
|
return;
|
|
|
|
}
|
|
|
|
|
2019-02-10 16:55:12 +00:00
|
|
|
UInt64 max_block_size = DEFAULT_BLOCK_SIZE;
|
2018-08-10 11:42:12 +00:00
|
|
|
if (params.has("max_block_size"))
|
|
|
|
{
|
|
|
|
std::string max_block_size_str = params.get("max_block_size", "");
|
|
|
|
if (max_block_size_str.empty())
|
|
|
|
{
|
2018-08-13 15:00:41 +00:00
|
|
|
process_error("Empty max_block_size specified");
|
2018-08-10 11:42:12 +00:00
|
|
|
return;
|
|
|
|
}
|
|
|
|
max_block_size = parse<size_t>(max_block_size_str);
|
|
|
|
}
|
2018-08-08 16:15:29 +00:00
|
|
|
|
2018-08-09 12:57:34 +00:00
|
|
|
std::string columns = params.get("columns");
|
|
|
|
std::unique_ptr<Block> sample_block;
|
2018-08-08 16:15:29 +00:00
|
|
|
try
|
|
|
|
{
|
2018-08-09 12:57:34 +00:00
|
|
|
sample_block = parseColumns(std::move(columns));
|
|
|
|
}
|
|
|
|
catch (const Exception & ex)
|
|
|
|
{
|
2018-08-13 15:00:41 +00:00
|
|
|
process_error("Invalid 'columns' parameter in request body '" + ex.message() + "'");
|
2020-01-02 06:56:53 +00:00
|
|
|
LOG_WARNING(log, ex.getStackTraceString());
|
2018-08-09 12:57:34 +00:00
|
|
|
return;
|
|
|
|
}
|
2018-08-08 16:15:29 +00:00
|
|
|
|
2018-08-09 12:57:34 +00:00
|
|
|
std::string format = params.get("format", "RowBinary");
|
|
|
|
std::string query = params.get("query");
|
2018-08-13 15:00:41 +00:00
|
|
|
LOG_TRACE(log, "Query: " << query);
|
2018-08-08 16:15:29 +00:00
|
|
|
|
2018-08-09 12:57:34 +00:00
|
|
|
std::string connection_string = params.get("connection_string");
|
2018-08-13 15:00:41 +00:00
|
|
|
LOG_TRACE(log, "Connection string: '" << connection_string << "'");
|
2018-08-08 16:15:29 +00:00
|
|
|
|
2018-08-09 12:57:34 +00:00
|
|
|
WriteBufferFromHTTPServerResponse out(request, response, keep_alive_timeout);
|
|
|
|
try
|
|
|
|
{
|
|
|
|
BlockOutputStreamPtr writer = FormatFactory::instance().getOutput(format, out, *sample_block, *context);
|
|
|
|
auto pool = getPool(connection_string);
|
|
|
|
ODBCBlockInputStream inp(pool->get(), query, *sample_block, max_block_size);
|
|
|
|
copyData(inp, *writer);
|
2018-08-08 16:15:29 +00:00
|
|
|
}
|
|
|
|
catch (...)
|
|
|
|
{
|
2018-08-13 15:00:41 +00:00
|
|
|
auto message = getCurrentExceptionMessage(true);
|
2018-08-08 16:15:29 +00:00
|
|
|
response.setStatusAndReason(
|
|
|
|
Poco::Net::HTTPResponse::HTTP_INTERNAL_SERVER_ERROR); // can't call process_error, bacause of too soon response sending
|
|
|
|
writeStringBinary(message, out);
|
2018-08-13 15:00:41 +00:00
|
|
|
tryLogCurrentException(log);
|
2018-08-08 16:15:29 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|