ClickHouse/programs/odbc-bridge/MainHandler.cpp

193 lines
5.9 KiB
C++
Raw Normal View History

#include "MainHandler.h"
#include "validateODBCConnectionString.h"
#include "ODBCBlockInputStream.h"
2020-04-28 00:56:44 +00:00
#include "ODBCBlockOutputStream.h"
2020-08-14 13:41:44 +00:00
#include "getIdentifierQuote.h"
#include <DataTypes/DataTypeFactory.h>
#include <Formats/FormatFactory.h>
#include <Server/HTTP/WriteBufferFromHTTPServerResponse.h>
#include <IO/WriteHelpers.h>
#include <IO/ReadHelpers.h>
2020-08-14 13:41:44 +00:00
#include <IO/ReadBufferFromIStream.h>
#include <Poco/Net/HTTPServerRequest.h>
#include <Poco/Net/HTTPServerResponse.h>
#include <Poco/Net/HTMLForm.h>
2020-08-14 13:41:44 +00:00
#include <Poco/ThreadPool.h>
2021-10-16 14:03:50 +00:00
#include <QueryPipeline/QueryPipeline.h>
2021-10-11 16:11:50 +00:00
#include <Processors/Executors/CompletedPipelineExecutor.h>
2021-10-15 10:15:14 +00:00
#include <Processors/Formats/IInputFormat.h>
2021-10-02 07:13:14 +00:00
#include <base/logger_useful.h>
#include <Server/HTTP/HTMLForm.h>
2020-08-14 13:41:44 +00:00
2019-04-17 17:36:58 +00:00
#include <mutex>
2020-08-14 13:41:44 +00:00
#include <memory>
namespace DB
{
2021-03-22 11:40:29 +00:00
namespace
{
std::unique_ptr<Block> parseColumns(std::string && column_string)
{
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;
}
}
void ODBCHandler::processError(HTTPServerResponse & response, const std::string & message)
2020-04-28 00:56:44 +00:00
{
response.setStatusAndReason(HTTPResponse::HTTP_INTERNAL_SERVER_ERROR);
2020-04-28 00:56:44 +00:00
if (!response.sent())
*response.send() << message << std::endl;
2020-05-23 22:24:01 +00:00
LOG_WARNING(log, message);
2020-04-28 00:56:44 +00:00
}
2021-03-22 11:40:29 +00:00
void ODBCHandler::handleRequest(HTTPServerRequest & request, HTTPServerResponse & response)
{
2021-06-16 14:33:14 +00:00
HTMLForm params(getContext()->getSettingsRef(), request);
2021-04-17 08:09:22 +00:00
LOG_TRACE(log, "Request URI: {}", request.getURI());
2020-04-28 00:56:44 +00:00
if (mode == "read")
params.read(request.getStream());
2020-05-14 21:51:07 +00:00
if (mode == "read" && !params.has("query"))
{
processError(response, "No 'query' in request body");
return;
}
2021-04-17 08:09:22 +00:00
if (!params.has("connection_string"))
{
2021-04-17 08:09:22 +00:00
processError(response, "No 'connection_string' in request URL");
return;
}
2021-04-17 08:09:22 +00:00
if (!params.has("sample_block"))
{
2021-04-17 08:09:22 +00:00
processError(response, "No 'sample_block' in request URL");
return;
}
2021-04-17 08:09:22 +00:00
std::string format = params.get("format", "RowBinary");
std::string connection_string = params.get("connection_string");
LOG_TRACE(log, "Connection string: '{}'", connection_string);
2020-05-14 21:51:07 +00:00
UInt64 max_block_size = DEFAULT_BLOCK_SIZE;
if (params.has("max_block_size"))
{
std::string max_block_size_str = params.get("max_block_size", "");
if (max_block_size_str.empty())
{
processError(response, "Empty max_block_size specified");
return;
}
max_block_size = parse<size_t>(max_block_size_str);
}
2021-04-17 08:09:22 +00:00
std::string sample_block_string = params.get("sample_block");
std::unique_ptr<Block> sample_block;
try
{
2021-04-17 08:09:22 +00:00
sample_block = parseColumns(std::move(sample_block_string));
}
catch (const Exception & ex)
{
2021-04-17 08:09:22 +00:00
processError(response, "Invalid 'sample_block' parameter in request body '" + ex.message() + "'");
LOG_ERROR(log, ex.getStackTraceString());
return;
}
WriteBufferFromHTTPServerResponse out(response, request.getMethod() == Poco::Net::HTTPRequest::HTTP_HEAD, keep_alive_timeout);
2020-05-14 21:51:07 +00:00
try
{
2021-06-07 18:09:16 +00:00
auto connection_handler = ODBCConnectionFactory::instance().get(
2021-04-06 18:59:34 +00:00
validateODBCConnectionString(connection_string),
getContext()->getSettingsRef().odbc_bridge_connection_pool_size);
2021-04-06 18:59:34 +00:00
2020-05-14 21:51:07 +00:00
if (mode == "write")
2020-04-28 00:56:44 +00:00
{
if (!params.has("db_name"))
{
processError(response, "No 'db_name' in request URL");
return;
}
if (!params.has("table_name"))
{
processError(response, "No 'table_name' in request URL");
return;
}
std::string db_name = params.get("db_name");
std::string table_name = params.get("table_name");
2020-05-23 22:24:01 +00:00
LOG_TRACE(log, "DB name: '{}', table name: '{}'", db_name, table_name);
2020-05-14 21:51:07 +00:00
2020-05-15 11:26:51 +00:00
auto quoting_style = IdentifierQuotingStyle::None;
#if USE_ODBC
2021-06-07 18:09:16 +00:00
quoting_style = getQuotingStyle(connection_handler);
2020-05-15 11:26:51 +00:00
#endif
auto & read_buf = request.getStream();
2021-10-11 16:11:50 +00:00
auto input_format = getContext()->getInputFormat(format, read_buf, *sample_block, max_block_size);
auto sink = std::make_shared<ODBCSink>(std::move(connection_handler), db_name, table_name, *sample_block, getContext(), quoting_style);
QueryPipeline pipeline(std::move(input_format));
pipeline.complete(std::move(sink));
CompletedPipelineExecutor executor(pipeline);
executor.execute();
2020-04-28 00:56:44 +00:00
writeStringBinary("Ok.", out);
}
2020-05-14 21:51:07 +00:00
else
2020-04-28 00:56:44 +00:00
{
2020-05-14 21:51:07 +00:00
std::string query = params.get("query");
2020-05-23 22:24:01 +00:00
LOG_TRACE(log, "Query: {}", query);
2020-04-28 00:56:44 +00:00
2021-10-11 16:11:50 +00:00
auto writer = FormatFactory::instance().getOutputFormatParallelIfPossible(format, out, *sample_block, getContext());
auto source = std::make_shared<ODBCSource>(std::move(connection_handler), query, *sample_block, max_block_size);
QueryPipeline pipeline(std::move(source));
pipeline.complete(std::move(writer));
CompletedPipelineExecutor executor(pipeline);
executor.execute();
2020-04-28 00:56:44 +00:00
}
2020-05-14 21:51:07 +00:00
}
catch (...)
{
auto message = getCurrentExceptionMessage(true);
response.setStatusAndReason(
Poco::Net::HTTPResponse::HTTP_INTERNAL_SERVER_ERROR); // can't call process_error, because of too soon response sending
try
{
writeStringBinary(message, out);
out.finalize();
}
catch (...)
{
tryLogCurrentException(log);
}
2020-05-14 21:51:07 +00:00
tryLogCurrentException(log);
}
2020-04-28 00:56:44 +00:00
try
{
out.finalize();
}
catch (...)
{
tryLogCurrentException(log);
}
}
2020-04-28 00:56:44 +00:00
}