ClickHouse/src/Storages/StorageURL.cpp

447 lines
15 KiB
C++
Raw Normal View History

2018-06-11 12:13:00 +00:00
#include <Storages/StorageURL.h>
#include <Interpreters/Context.h>
#include <Interpreters/evaluateConstantExpression.h>
#include <Parsers/ASTCreateQuery.h>
2018-06-11 12:13:00 +00:00
#include <Parsers/ASTLiteral.h>
2019-11-20 14:48:01 +00:00
#include <IO/ReadHelpers.h>
#include <IO/WriteBufferFromHTTP.h>
2019-11-20 14:48:01 +00:00
#include <IO/WriteHelpers.h>
#include <IO/ConnectionTimeouts.h>
#include <IO/ConnectionTimeoutsContext.h>
2018-06-11 12:13:00 +00:00
#include <Formats/FormatFactory.h>
2020-05-18 10:00:22 +00:00
#include <Processors/Formats/InputStreamFromInputFormat.h>
#include <DataStreams/IBlockOutputStream.h>
2021-07-21 16:13:17 +00:00
#include <Processors/Transforms/AddingDefaultsTransform.h>
2018-06-11 12:13:00 +00:00
#include <Poco/Net/HTTPRequest.h>
#include <Processors/Sources/SourceWithProgress.h>
2021-07-20 18:18:43 +00:00
#include <Processors/QueryPipeline.h>
#include <Processors/Executors/PullingPipelineExecutor.h>
2021-04-21 12:32:57 +00:00
#include <common/logger_useful.h>
2021-04-21 14:36:04 +00:00
#include <algorithm>
2018-06-11 12:13:00 +00:00
namespace DB
{
namespace ErrorCodes
{
extern const int NUMBER_OF_ARGUMENTS_DOESNT_MATCH;
2021-04-21 14:36:04 +00:00
extern const int NETWORK_ERROR;
}
2018-06-11 12:13:00 +00:00
2019-08-24 21:20:20 +00:00
IStorageURLBase::IStorageURLBase(
const Poco::URI & uri_,
ContextPtr /*context_*/,
2019-12-04 16:06:55 +00:00
const StorageID & table_id_,
2018-06-11 12:13:00 +00:00
const String & format_name_,
const std::optional<FormatSettings> & format_settings_,
2019-08-24 21:20:20 +00:00
const ColumnsDescription & columns_,
const ConstraintsDescription & constraints_,
2021-04-23 12:18:23 +00:00
const String & comment,
2021-09-07 11:17:25 +00:00
const String & compression_method_,
const ReadWriteBufferFromHTTP::HTTPHeaderEntries & headers_)
: IStorage(table_id_), uri(uri_), compression_method(compression_method_), format_name(format_name_), format_settings(format_settings_), headers(headers_)
2018-06-11 12:13:00 +00:00
{
2020-06-19 15:39:41 +00:00
StorageInMemoryMetadata storage_metadata;
storage_metadata.setColumns(columns_);
storage_metadata.setConstraints(constraints_);
2021-04-23 12:18:23 +00:00
storage_metadata.setComment(comment);
2020-06-19 15:39:41 +00:00
setInMemoryMetadata(storage_metadata);
2018-06-11 12:13:00 +00:00
}
namespace
{
class StorageURLSource : public SourceWithProgress
2018-06-11 12:13:00 +00:00
{
public:
StorageURLSource(const Poco::URI & uri,
const std::string & method,
std::function<void(std::ostream &)> callback,
2018-06-11 12:13:00 +00:00
const String & format,
const std::optional<FormatSettings> & format_settings,
String name_,
2018-06-11 12:13:00 +00:00
const Block & sample_block,
ContextPtr context,
2020-10-02 12:38:50 +00:00
const ColumnsDescription & columns,
2019-02-10 16:55:12 +00:00
UInt64 max_block_size,
const ConnectionTimeouts & timeouts,
2021-09-07 11:17:25 +00:00
const CompressionMethod compression_method,
const ReadWriteBufferFromHTTP::HTTPHeaderEntries & headers_ = {})
: SourceWithProgress(sample_block), name(std::move(name_))
2018-06-11 12:13:00 +00:00
{
2021-09-07 11:17:25 +00:00
ReadWriteBufferFromHTTP::HTTPHeaderEntries headers;
for (const auto & header : headers_)
headers.emplace_back(header);
2020-08-28 01:21:08 +00:00
// Propagate OpenTelemetry trace context, if any, downstream.
2020-11-18 17:43:18 +00:00
if (CurrentThread::isInitialized())
2020-08-28 01:21:08 +00:00
{
2020-11-18 17:43:18 +00:00
const auto & thread_trace_context = CurrentThread::get().thread_trace_context;
2021-05-04 22:42:14 +00:00
if (thread_trace_context.trace_id != UUID())
2020-08-28 01:21:08 +00:00
{
2021-09-07 11:17:25 +00:00
headers.emplace_back("traceparent",
2020-11-18 17:43:18 +00:00
thread_trace_context.composeTraceparentHeader());
if (!thread_trace_context.tracestate.empty())
{
2021-09-07 11:17:25 +00:00
headers.emplace_back("tracestate",
2020-11-18 17:43:18 +00:00
thread_trace_context.tracestate);
}
2020-08-28 01:21:08 +00:00
}
}
read_buf = wrapReadBufferWithCompressionMethod(
std::make_unique<ReadWriteBufferFromHTTP>(
uri,
method,
std::move(callback),
timeouts,
context->getSettingsRef().max_http_get_redirects,
Poco::Net::HTTPBasicCredentials{},
DBMS_DEFAULT_BUFFER_SIZE,
2021-09-07 11:17:25 +00:00
headers,
context->getRemoteHostFilter()),
compression_method);
2019-12-03 17:15:16 +00:00
2020-12-07 22:52:51 +00:00
auto input_format = FormatFactory::instance().getInput(format, *read_buf, sample_block, context, max_block_size, format_settings);
2021-07-20 18:18:43 +00:00
pipeline = std::make_unique<QueryPipeline>();
pipeline->init(Pipe(input_format));
pipeline->addSimpleTransform([&](const Block & cur_header)
{
return std::make_shared<AddingDefaultsTransform>(cur_header, columns, *input_format, context);
});
reader = std::make_unique<PullingPipelineExecutor>(*pipeline);
2018-06-11 12:13:00 +00:00
}
String getName() const override
{
return name;
}
Chunk generate() override
2018-06-11 12:13:00 +00:00
{
if (!reader)
return {};
2018-06-11 12:13:00 +00:00
2021-07-20 18:18:43 +00:00
Chunk chunk;
if (reader->pull(chunk))
return chunk;
2018-06-11 12:13:00 +00:00
2021-07-20 18:18:43 +00:00
pipeline->reset();
reader.reset();
return {};
2018-06-11 12:13:00 +00:00
}
private:
String name;
std::unique_ptr<ReadBuffer> read_buf;
2021-07-20 18:18:43 +00:00
std::unique_ptr<QueryPipeline> pipeline;
std::unique_ptr<PullingPipelineExecutor> reader;
2018-06-11 12:13:00 +00:00
};
}
2018-06-16 05:54:06 +00:00
2021-07-23 14:25:35 +00:00
StorageURLSink::StorageURLSink(
const Poco::URI & uri,
const String & format,
const std::optional<FormatSettings> & format_settings,
const Block & sample_block,
ContextPtr context,
const ConnectionTimeouts & timeouts,
const CompressionMethod compression_method)
: SinkToStorage(sample_block)
2020-04-28 00:56:44 +00:00
{
write_buf = wrapWriteBufferWithCompressionMethod(
std::make_unique<WriteBufferFromHTTP>(uri, Poco::Net::HTTPRequest::HTTP_POST, timeouts),
compression_method, 3);
2020-12-30 03:07:30 +00:00
writer = FormatFactory::instance().getOutputStream(format, *write_buf, sample_block,
context, {} /* write callback */, format_settings);
2020-04-28 00:56:44 +00:00
}
2018-06-16 05:54:06 +00:00
2020-07-09 01:00:16 +00:00
2021-07-23 14:25:35 +00:00
void StorageURLSink::consume(Chunk chunk)
2020-07-09 01:00:16 +00:00
{
2021-07-23 14:25:35 +00:00
if (is_first_chunk)
{
writer->writePrefix();
is_first_chunk = false;
}
2020-07-09 01:00:16 +00:00
2021-07-23 14:25:35 +00:00
writer->write(getPort().getHeader().cloneWithColumns(chunk.detachColumns()));
2020-07-09 01:00:16 +00:00
}
2021-07-23 14:25:35 +00:00
void StorageURLSink::onFinish()
2020-07-09 01:00:16 +00:00
{
writer->writeSuffix();
writer->flush();
write_buf->finalize();
}
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 StorageMetadataPtr & /*metadata_snapshot*/,
2018-06-11 12:13:00 +00:00
const SelectQueryInfo & /*query_info*/,
ContextPtr /*context*/,
2018-06-11 12:13:00 +00:00
QueryProcessingStage::Enum & /*processed_stage*/,
size_t /*max_block_size*/) const
{
return {};
}
std::function<void(std::ostream &)> IStorageURLBase::getReadPOSTDataCallback(
const Names & /*column_names*/,
const StorageMetadataPtr & /*metadata_snapshot*/,
2018-06-11 12:13:00 +00:00
const SelectQueryInfo & /*query_info*/,
ContextPtr /*context*/,
QueryProcessingStage::Enum & /*processed_stage*/,
size_t /*max_block_size*/) const
{
return nullptr;
}
2020-08-03 13:54:14 +00:00
Pipe IStorageURLBase::read(
const Names & column_names,
const StorageMetadataPtr & metadata_snapshot,
SelectQueryInfo & query_info,
ContextPtr local_context,
2018-06-25 12:21:54 +00:00
QueryProcessingStage::Enum processed_stage,
size_t max_block_size,
2018-06-11 12:13:00 +00:00
unsigned /*num_streams*/)
{
auto request_uri = uri;
auto params = getReadURIParams(column_names, metadata_snapshot, query_info, local_context, processed_stage, max_block_size);
for (const auto & [param, value] : params)
request_uri.addQueryParameter(param, value);
2020-08-03 13:54:14 +00:00
return Pipe(std::make_shared<StorageURLSource>(
request_uri,
getReadMethod(),
getReadPOSTDataCallback(
column_names, metadata_snapshot, query_info,
local_context, processed_stage, max_block_size),
2018-06-11 12:13:00 +00:00
format_name,
format_settings,
2018-06-11 12:13:00 +00:00
getName(),
getHeaderBlock(column_names, metadata_snapshot),
local_context,
2020-10-02 12:38:50 +00:00
metadata_snapshot->getColumns(),
2018-06-11 12:13:00 +00:00
max_block_size,
ConnectionTimeouts::getHTTPTimeouts(local_context),
2021-09-07 11:17:25 +00:00
chooseCompressionMethod(request_uri.getPath(), compression_method),
headers));
2018-06-11 12:13:00 +00:00
}
2021-04-21 14:36:04 +00:00
Pipe StorageURLWithFailover::read(
const Names & column_names,
const StorageMetadataPtr & metadata_snapshot,
SelectQueryInfo & query_info,
ContextPtr local_context,
QueryProcessingStage::Enum processed_stage,
size_t max_block_size,
unsigned /*num_streams*/)
{
auto params = getReadURIParams(column_names, metadata_snapshot, query_info, local_context, processed_stage, max_block_size);
2021-04-27 08:49:24 +00:00
WriteBufferFromOwnString error_message;
error_message << "Detailed description:";
2021-04-21 14:36:04 +00:00
for (const auto & uri_option : uri_options)
{
auto request_uri = uri_option;
for (const auto & [param, value] : params)
request_uri.addQueryParameter(param, value);
try
{
2021-04-21 17:09:01 +00:00
/// Check for uri accessibility is done in constructor of ReadWriteBufferFromHTTP while creating StorageURLSource.
2021-04-21 14:36:04 +00:00
auto url_source = std::make_shared<StorageURLSource>(
request_uri,
getReadMethod(),
getReadPOSTDataCallback(
column_names, metadata_snapshot, query_info,
local_context, processed_stage, max_block_size),
format_name,
format_settings,
getName(),
getHeaderBlock(column_names, metadata_snapshot),
local_context,
metadata_snapshot->getColumns(),
max_block_size,
ConnectionTimeouts::getHTTPTimeouts(local_context),
chooseCompressionMethod(request_uri.getPath(), compression_method));
2021-04-21 19:53:38 +00:00
std::shuffle(uri_options.begin(), uri_options.end(), thread_local_rng);
2021-04-21 14:36:04 +00:00
return Pipe(url_source);
}
catch (...)
{
2021-04-27 08:49:24 +00:00
error_message << " Host: " << uri_option.getHost() << ", post: " << uri_option.getPort() << ", path: " << uri_option.getPath();
error_message << ", error: " << getCurrentExceptionMessage(false) << ";";
2021-04-21 14:36:04 +00:00
tryLogCurrentException(__PRETTY_FUNCTION__);
}
}
2021-04-27 08:49:24 +00:00
throw Exception(ErrorCodes::NETWORK_ERROR, "All uri options are unreachable. {}", error_message.str());
2021-04-21 14:36:04 +00:00
}
2021-07-23 14:25:35 +00:00
SinkToStoragePtr IStorageURLBase::write(const ASTPtr & /*query*/, const StorageMetadataPtr & metadata_snapshot, ContextPtr context)
2018-06-11 12:13:00 +00:00
{
2021-07-23 14:25:35 +00:00
return std::make_shared<StorageURLSink>(uri, format_name,
format_settings, metadata_snapshot->getSampleBlock(), context,
ConnectionTimeouts::getHTTPTimeouts(context),
chooseCompressionMethod(uri.toString(), compression_method));
2018-06-11 12:13:00 +00:00
}
2018-06-16 05:54:06 +00:00
2021-04-23 12:18:23 +00:00
StorageURL::StorageURL(
const Poco::URI & uri_,
const StorageID & table_id_,
const String & format_name_,
const std::optional<FormatSettings> & format_settings_,
const ColumnsDescription & columns_,
const ConstraintsDescription & constraints_,
const String & comment,
ContextPtr context_,
2021-09-07 11:17:25 +00:00
const String & compression_method_,
const ReadWriteBufferFromHTTP::HTTPHeaderEntries & headers_)
: IStorageURLBase(uri_, context_, table_id_, format_name_, format_settings_, columns_, constraints_, comment, compression_method_, headers_)
2021-03-23 11:29:29 +00:00
{
context_->getRemoteHostFilter().checkURL(uri);
2021-03-23 11:29:29 +00:00
}
2021-04-21 12:32:57 +00:00
2021-04-21 14:36:04 +00:00
StorageURLWithFailover::StorageURLWithFailover(
2021-05-02 16:33:45 +00:00
const std::vector<String> & uri_options_,
const StorageID & table_id_,
const String & format_name_,
const std::optional<FormatSettings> & format_settings_,
const ColumnsDescription & columns_,
const ConstraintsDescription & constraints_,
ContextPtr context_,
const String & compression_method_)
: StorageURL(Poco::URI(), table_id_, format_name_, format_settings_, columns_, constraints_, String{}, context_, compression_method_)
2021-04-21 14:36:04 +00:00
{
for (const auto & uri_option : uri_options_)
{
2021-04-22 07:37:20 +00:00
Poco::URI poco_uri(uri_option);
context_->getRemoteHostFilter().checkURL(poco_uri);
uri_options.emplace_back(std::move(poco_uri));
2021-04-21 14:36:04 +00:00
LOG_DEBUG(&Poco::Logger::get("StorageURLDistributed"), "Adding URL option: {}", uri_option);
}
}
2021-04-21 12:32:57 +00:00
FormatSettings StorageURL::getFormatSettingsFromArgs(const StorageFactory::Arguments & args)
{
// Use format settings from global server context + settings from
// the SETTINGS clause of the create query. Settings from current
// session and user are ignored.
FormatSettings format_settings;
if (args.storage_def->settings)
{
FormatFactorySettings user_format_settings;
// Apply changed settings from global context, but ignore the
// unknown ones, because we only have the format settings here.
const auto & changes = args.getContext()->getSettingsRef().changes();
for (const auto & change : changes)
{
if (user_format_settings.has(change.name))
{
user_format_settings.set(change.name, change.value);
}
}
// Apply changes from SETTINGS clause, with validation.
user_format_settings.applyChanges(args.storage_def->settings->changes);
format_settings = getFormatSettings(args.getContext(),
user_format_settings);
}
else
{
format_settings = getFormatSettings(args.getContext());
}
return format_settings;
}
2021-09-07 11:17:25 +00:00
URLBasedDataSourceConfiguration StorageURL::getConfiguration(ASTs & args, ContextPtr local_context)
2018-06-11 12:13:00 +00:00
{
2021-09-07 11:17:25 +00:00
auto [common_configuration, storage_specific_args, with_named_collection] = getURLBasedDataSourceConfiguration(args, local_context);
URLBasedDataSourceConfiguration configuration(common_configuration);
2018-06-11 12:13:00 +00:00
2021-09-07 11:17:25 +00:00
if (with_named_collection)
{
if (!storage_specific_args.empty())
throw Exception(ErrorCodes::BAD_ARGUMENTS, "Illegal arguments");
}
else
{
if (args.size() != 2 && args.size() != 3)
2018-06-11 12:13:00 +00:00
throw Exception(
"Storage URL requires 2 or 3 arguments: url, name of used format and optional compression method.", ErrorCodes::NUMBER_OF_ARGUMENTS_DOESNT_MATCH);
2018-06-11 12:13:00 +00:00
2021-09-07 11:17:25 +00:00
for (auto & arg : args)
arg = evaluateConstantExpressionOrIdentifierAsLiteral(arg, local_context);
2018-06-11 12:13:00 +00:00
2021-09-07 11:17:25 +00:00
configuration.url = args[0]->as<ASTLiteral &>().value.safeGet<String>();
configuration.format = args[1]->as<ASTLiteral &>().value.safeGet<String>();
if (args.size() == 3)
configuration.compression_method = args[2]->as<ASTLiteral &>().value.safeGet<String>();
}
2018-06-11 12:13:00 +00:00
2021-09-07 11:17:25 +00:00
return configuration;
}
2018-06-11 12:13:00 +00:00
2021-09-07 11:17:25 +00:00
void registerStorageURL(StorageFactory & factory)
{
factory.registerStorage("URL", [](const StorageFactory::Arguments & args)
{
ASTs & engine_args = args.engine_args;
auto configuration = StorageURL::getConfiguration(engine_args, args.getContext());
auto format_settings = StorageURL::getFormatSettingsFromArgs(args);
Poco::URI uri(configuration.url);
ReadWriteBufferFromHTTP::HTTPHeaderEntries headers;
for (const auto & [header, value] : configuration.headers)
{
2021-09-07 11:17:25 +00:00
auto value_literal = value.safeGet<String>();
headers.emplace_back(std::make_pair(header, value_literal));
}
return StorageURL::create(
uri,
2019-12-04 16:06:55 +00:00
args.table_id,
2021-09-07 11:17:25 +00:00
configuration.format,
format_settings,
2021-04-23 12:18:23 +00:00
args.columns,
args.constraints,
args.comment,
args.getContext(),
2021-09-07 11:17:25 +00:00
configuration.compression_method,
headers);
},
{
.supports_settings = true,
.source_access_type = AccessType::URL,
2018-06-11 12:13:00 +00:00
});
}
}