ClickHouse/src/TableFunctions/TableFunctionS3.cpp

197 lines
7.3 KiB
C++
Raw Normal View History

#include "config.h"
2019-12-06 14:37:21 +00:00
#if USE_AWS_S3
2019-12-03 16:23:24 +00:00
#include <IO/S3Common.h>
#include <Interpreters/evaluateConstantExpression.h>
#include <Interpreters/Context.h>
2019-05-31 07:27:14 +00:00
#include <TableFunctions/TableFunctionFactory.h>
#include <TableFunctions/TableFunctionS3.h>
#include <Interpreters/parseColumnsListForTableFunction.h>
#include <Access/Common/AccessFlags.h>
#include <Parsers/ASTLiteral.h>
#include <Storages/checkAndGetLiteralArgument.h>
2021-09-07 11:17:25 +00:00
#include <Storages/StorageS3.h>
2022-09-13 13:07:43 +00:00
#include <Storages/StorageURL.h>
#include <Storages/NamedCollectionsHelpers.h>
#include <Formats/FormatFactory.h>
2019-12-15 06:34:43 +00:00
#include "registerTableFunctions.h"
2022-03-28 19:18:20 +00:00
#include <filesystem>
2019-05-31 07:27:14 +00:00
2019-05-31 07:27:14 +00:00
namespace DB
{
2019-09-22 22:13:42 +00:00
namespace ErrorCodes
{
extern const int NUMBER_OF_ARGUMENTS_DOESNT_MATCH;
}
2021-09-07 11:17:25 +00:00
/// This is needed to avoid copy-pase. Because s3Cluster arguments only differ in additional argument (first) - cluster name
void TableFunctionS3::parseArgumentsImpl(const String & error_message, ASTs & args, ContextPtr context, StorageS3::Configuration & s3_configuration)
{
if (auto named_collection = tryGetNamedCollectionWithOverrides(args))
2021-09-07 11:17:25 +00:00
{
StorageS3::processNamedCollectionResult(s3_configuration, *named_collection);
2021-09-07 11:17:25 +00:00
}
else
{
2022-02-06 17:07:35 +00:00
if (args.empty() || args.size() > 6)
2023-01-26 09:52:47 +00:00
throw Exception::createDeprecated(error_message, ErrorCodes::NUMBER_OF_ARGUMENTS_DOESNT_MATCH);
auto * header_it = StorageURL::collectHeaders(args, s3_configuration.headers_from_ast, context);
2022-09-13 13:07:43 +00:00
if (header_it != args.end())
args.erase(header_it);
2021-09-07 11:17:25 +00:00
for (auto & arg : args)
arg = evaluateConstantExpressionOrIdentifierAsLiteral(arg, context);
2021-09-07 11:17:25 +00:00
/// Size -> argument indexes
static auto size_to_args = std::map<size_t, std::map<String, size_t>>
{
{1, {{}}},
{2, {{"format", 1}}},
2021-09-07 11:17:25 +00:00
{5, {{"access_key_id", 1}, {"secret_access_key", 2}, {"format", 3}, {"structure", 4}}},
{6, {{"access_key_id", 1}, {"secret_access_key", 2}, {"format", 3}, {"structure", 4}, {"compression_method", 5}}}
};
std::map<String, size_t> args_to_idx;
/// For 4 arguments we support 2 possible variants:
/// s3(source, format, structure, compression_method) and s3(source, access_key_id, access_key_id, format)
/// We can distinguish them by looking at the 2-nd argument: check if it's a format name or not.
if (args.size() == 4)
{
auto second_arg = checkAndGetLiteralArgument<String>(args[1], "format/access_key_id");
if (second_arg == "auto" || FormatFactory::instance().getAllFormats().contains(second_arg))
args_to_idx = {{"format", 1}, {"structure", 2}, {"compression_method", 3}};
2022-02-15 11:57:38 +00:00
else
args_to_idx = {{"access_key_id", 1}, {"secret_access_key", 2}, {"format", 3}};
}
/// For 3 arguments we support 2 possible variants:
/// s3(source, format, structure) and s3(source, access_key_id, access_key_id)
/// We can distinguish them by looking at the 2-nd argument: check if it's a format name or not.
else if (args.size() == 3)
{
auto second_arg = checkAndGetLiteralArgument<String>(args[1], "format/access_key_id");
if (second_arg == "auto" || FormatFactory::instance().getAllFormats().contains(second_arg))
args_to_idx = {{"format", 1}, {"structure", 2}};
else
args_to_idx = {{"access_key_id", 1}, {"secret_access_key", 2}};
}
else
{
args_to_idx = size_to_args[args.size()];
}
2021-09-07 11:17:25 +00:00
/// This argument is always the first
s3_configuration.url = S3::URI(checkAndGetLiteralArgument<String>(args[0], "url"));
2021-09-07 11:17:25 +00:00
if (args_to_idx.contains("format"))
s3_configuration.format = checkAndGetLiteralArgument<String>(args[args_to_idx["format"]], "format");
2021-09-07 11:17:25 +00:00
if (args_to_idx.contains("structure"))
s3_configuration.structure = checkAndGetLiteralArgument<String>(args[args_to_idx["structure"]], "structure");
2021-09-07 11:17:25 +00:00
if (args_to_idx.contains("compression_method"))
s3_configuration.compression_method = checkAndGetLiteralArgument<String>(args[args_to_idx["compression_method"]], "compression_method");
2021-09-07 11:17:25 +00:00
if (args_to_idx.contains("access_key_id"))
s3_configuration.auth_settings.access_key_id = checkAndGetLiteralArgument<String>(args[args_to_idx["access_key_id"]], "access_key_id");
2021-09-07 11:17:25 +00:00
if (args_to_idx.contains("secret_access_key"))
s3_configuration.auth_settings.secret_access_key = checkAndGetLiteralArgument<String>(args[args_to_idx["secret_access_key"]], "secret_access_key");
2021-09-07 11:17:25 +00:00
}
if (s3_configuration.format == "auto")
s3_configuration.format = FormatFactory::instance().getFormatFromFileName(s3_configuration.url.uri.getPath(), true);
}
void TableFunctionS3::parseArguments(const ASTPtr & ast_function, ContextPtr context)
{
/// Parse args
ASTs & args_func = ast_function->children;
const auto message = fmt::format(
"The signature of table function {} could be the following:\n" \
" - url\n" \
" - url, format\n" \
" - url, format, structure\n" \
" - url, access_key_id, secret_access_key\n" \
" - url, format, structure, compression_method\n" \
" - url, access_key_id, secret_access_key, format\n" \
" - url, access_key_id, secret_access_key, format, structure\n" \
" - url, access_key_id, secret_access_key, format, structure, compression_method",
getName());
if (args_func.size() != 1)
throw Exception(ErrorCodes::NUMBER_OF_ARGUMENTS_DOESNT_MATCH, "Table function '{}' must have arguments.", getName());
auto & args = args_func.at(0)->children;
parseArgumentsImpl(message, args, context, configuration);
}
ColumnsDescription TableFunctionS3::getActualTableStructure(ContextPtr context) const
{
if (configuration.structure == "auto")
{
context->checkAccess(getSourceAccessType());
2022-09-13 13:07:43 +00:00
return StorageS3::getTableStructureFromData(configuration, false, std::nullopt, context);
}
return parseColumnsListFromString(configuration.structure, context);
}
bool TableFunctionS3::supportsReadingSubsetOfColumns()
{
return FormatFactory::instance().checkIfFormatSupportsSubsetOfColumns(configuration.format);
}
StoragePtr TableFunctionS3::executeImpl(const ASTPtr & /*ast_function*/, ContextPtr context, const std::string & table_name, ColumnsDescription /*cached_columns*/) const
{
2022-11-18 04:09:11 +00:00
S3::URI s3_uri (configuration.url);
ColumnsDescription columns;
if (configuration.structure != "auto")
columns = parseColumnsListFromString(configuration.structure, context);
else if (!structure_hint.empty())
columns = structure_hint;
StoragePtr storage = std::make_shared<StorageS3>(
2022-09-13 13:07:43 +00:00
configuration,
2021-04-23 12:18:23 +00:00
StorageID(getDatabaseName(), table_name),
columns,
2021-04-23 12:18:23 +00:00
ConstraintsDescription{},
String{},
context,
2021-08-23 19:05:28 +00:00
/// No format_settings for table function S3
2022-09-13 13:07:43 +00:00
std::nullopt);
storage->startup();
return storage;
}
2019-05-31 07:27:14 +00:00
void registerTableFunctionS3(TableFunctionFactory & factory)
{
factory.registerFunction<TableFunctionS3>();
}
2019-09-22 22:13:42 +00:00
void registerTableFunctionCOS(TableFunctionFactory & factory)
{
2020-07-17 03:33:29 +00:00
factory.registerFunction<TableFunctionCOS>();
}
2022-11-11 08:40:10 +00:00
void registerTableFunctionOSS(TableFunctionFactory & factory)
{
factory.registerFunction<TableFunctionOSS>();
}
2019-05-31 07:27:14 +00:00
}
2019-12-06 14:37:21 +00:00
#endif