thevar1able/ClickHouse
1
0
Fork 0
mirror of https://github.com/ClickHouse/ClickHouse.git synced 2024-11-21 15:12:02 +00:00
Code Issues Packages Projects Releases Wiki Activity

Better

Browse Source
This commit is contained in:
kssenii 2021-04-20 07:53:55 +00:00
parent bb767b66d5
commit da7a4ac1ea
6 changed files with 72 additions and 69 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
src/Storages/HDFS/HDFSCommon.cpp
View File

@ -9,14 +9,15 @@
#include <IO/Operators.h>
#include <common/logger_useful.h>
namespace DB
{
namespace ErrorCodes
{
extern const int BAD_ARGUMENTS;
extern const int NETWORK_ERROR;
extern const int EXCESSIVE_ELEMENT_IN_CONFIG;
extern const int NO_ELEMENTS_IN_CONFIG;
extern const int BAD_ARGUMENTS;
extern const int NETWORK_ERROR;
extern const int EXCESSIVE_ELEMENT_IN_CONFIG;
extern const int NO_ELEMENTS_IN_CONFIG;
}
const String HDFSBuilderWrapper::CONFIG_PREFIX = "hdfs";

80
src/Storages/HDFS/HDFSCommon.h
View File

@ -17,6 +17,7 @@
namespace DB
{
namespace detail
{
struct HDFSFsDeleter
@ -28,16 +29,14 @@ namespace detail
};
}
struct HDFSFileInfo
{
hdfsFileInfo * file_info;
int length;
HDFSFileInfo()
: file_info(nullptr)
, length(0)
{
}
HDFSFileInfo() : file_info(nullptr) , length(0) {}
HDFSFileInfo(const HDFSFileInfo & other) = delete;
HDFSFileInfo(HDFSFileInfo && other) = default;
HDFSFileInfo & operator=(const HDFSFileInfo & other) = delete;
@ -49,17 +48,30 @@ struct HDFSFileInfo
}
};
class HDFSBuilderWrapper
{
hdfsBuilder * hdfs_builder;
String hadoop_kerberos_keytab;
String hadoop_kerberos_principal;
String hadoop_kerberos_kinit_command = "kinit";
String hadoop_security_kerberos_ticket_cache_path;
static std::mutex kinit_mtx;
friend HDFSBuilderWrapper createHDFSBuilder(const String & uri_str, const Poco::Util::AbstractConfiguration &);
std::vector<std::pair<String, String>> config_stor;
static const String CONFIG_PREFIX;
public:
HDFSBuilderWrapper() : hdfs_builder(hdfsNewBuilder()) {}
~HDFSBuilderWrapper() { hdfsFreeBuilder(hdfs_builder); }
HDFSBuilderWrapper(const HDFSBuilderWrapper &) = delete;
HDFSBuilderWrapper(HDFSBuilderWrapper &&) = default;
hdfsBuilder * get() { return hdfs_builder; }
private:
void loadFromConfig(const Poco::Util::AbstractConfiguration & config, const String & config_path, bool isUser = false);
String getKinitCmd();
void runKinit();
// hdfs builder relies on an external config data storage
std::pair<String, String>& keep(const String & k, const String & v)
@ -67,48 +79,24 @@ class HDFSBuilderWrapper
return config_stor.emplace_back(std::make_pair(k, v));
}
hdfsBuilder * hdfs_builder;
String hadoop_kerberos_keytab;
String hadoop_kerberos_principal;
String hadoop_kerberos_kinit_command = "kinit";
String hadoop_security_kerberos_ticket_cache_path;
static std::mutex kinit_mtx;
std::vector<std::pair<String, String>> config_stor;
bool need_kinit{false};
static const String CONFIG_PREFIX;
private:
void loadFromConfig(const Poco::Util::AbstractConfiguration & config, const String & config_path, bool isUser = false);
String getKinitCmd();
void runKinit();
public:
hdfsBuilder *
get()
{
return hdfs_builder;
}
HDFSBuilderWrapper()
: hdfs_builder(hdfsNewBuilder())
{
}
~HDFSBuilderWrapper()
{
hdfsFreeBuilder(hdfs_builder);
}
HDFSBuilderWrapper(const HDFSBuilderWrapper &) = delete;
HDFSBuilderWrapper(HDFSBuilderWrapper &&) = default;
friend HDFSBuilderWrapper createHDFSBuilder(const String & uri_str, const Poco::Util::AbstractConfiguration &);
};
using HDFSFSPtr = std::unique_ptr<std::remove_pointer_t<hdfsFS>, detail::HDFSFsDeleter>;
// set read/connect timeout, default value in libhdfs3 is about 1 hour, and too large
/// TODO Allow to tune from query Settings.
HDFSBuilderWrapper createHDFSBuilder(const String & uri_str, const Poco::Util::AbstractConfiguration &);
HDFSFSPtr createHDFSFS(hdfsBuilder * builder);
}
#endif

20
src/Storages/HDFS/ReadBufferFromHDFS.cpp
View File

@ -8,6 +8,7 @@
namespace DB
{
namespace ErrorCodes
{
extern const int NETWORK_ERROR;
@ -21,8 +22,9 @@ struct ReadBufferFromHDFS::ReadBufferFromHDFSImpl
/// HDFS create/open functions are not thread safe
static std::mutex hdfs_init_mutex;
std::string hdfs_uri;
std::string hdfs_file_path;
String hdfs_uri;
String hdfs_file_path;
hdfsFile fin;
HDFSBuilderWrapper builder;
HDFSFSPtr fs;
@ -66,16 +68,14 @@ struct ReadBufferFromHDFS::ReadBufferFromHDFSImpl
std::mutex ReadBufferFromHDFS::ReadBufferFromHDFSImpl::hdfs_init_mutex;
ReadBufferFromHDFS::ReadBufferFromHDFS(const std::string & hdfs_name_,
const Poco::Util::AbstractConfiguration & config_,
size_t buf_size_)
ReadBufferFromHDFS::ReadBufferFromHDFS(
const String & hdfs_uri_,
const String & hdfs_file_path_,
const Poco::Util::AbstractConfiguration & config_,
size_t buf_size_)
: BufferWithOwnMemory<ReadBuffer>(buf_size_)
, impl(std::make_unique<ReadBufferFromHDFSImpl>(hdfs_uri_, hdfs_file_path_, config_))
{
const size_t begin_of_path = hdfs_name_.find('/', hdfs_name_.find("//") + 2);
const String hdfs_file_path = hdfs_name_.substr(begin_of_path);
const String hdfs_uri = hdfs_name_.substr(0, begin_of_path) + "/";
impl = std::make_unique<ReadBufferFromHDFSImpl>(hdfs_uri, hdfs_file_path, config_);
}

15
src/Storages/HDFS/ReadBufferFromHDFS.h
View File

@ -7,11 +7,8 @@
#include <IO/BufferWithOwnMemory.h>
#include <string>
#include <memory>
#include <hdfs/hdfs.h>
#include <common/types.h>
#include <Interpreters/Context.h>
@ -22,13 +19,19 @@ namespace DB
*/
class ReadBufferFromHDFS : public BufferWithOwnMemory<ReadBuffer>
{
struct ReadBufferFromHDFSImpl;
std::unique_ptr<ReadBufferFromHDFSImpl> impl;
struct ReadBufferFromHDFSImpl;
public:
ReadBufferFromHDFS(const std::string & hdfs_name_, const Poco::Util::AbstractConfiguration &, size_t buf_size_ = DBMS_DEFAULT_BUFFER_SIZE);
ReadBufferFromHDFS(const String & hdfs_uri_, const String & hdfs_file_path_,
const Poco::Util::AbstractConfiguration &, size_t buf_size_ = DBMS_DEFAULT_BUFFER_SIZE);
~ReadBufferFromHDFS() override;
bool nextImpl() override;
private:
std::unique_ptr<ReadBufferFromHDFSImpl> impl;
};
}
#endif

15
src/Storages/HDFS/StorageHDFS.cpp
View File

@ -122,7 +122,7 @@ public:
current_path = uri + path;
auto compression = chooseCompressionMethod(path, compression_method);
auto read_buf = wrapReadBufferWithCompressionMethod(std::make_unique<ReadBufferFromHDFS>(current_path, getContext()->getGlobalContext()->getConfigRef()), compression);
auto read_buf = wrapReadBufferWithCompressionMethod(std::make_unique<ReadBufferFromHDFS>(uri, path, getContext()->getGlobalContext()->getConfigRef()), compression);
auto input_format = FormatFactory::instance().getInput(format, *read_buf, sample_block, getContext(), max_block_size);
auto input_stream = std::make_shared<InputStreamFromInputFormat>(input_format);
@ -271,7 +271,15 @@ Pipe StorageHDFS::read(
size_t max_block_size,
unsigned num_streams)
{
const size_t begin_of_path = uri.find('/', uri.find("//") + 2);
size_t begin_of_path;
/// This uri is checked for correctness in constructor of StorageHDFS and never modified afterwards
auto two_slash = uri.find("//");
if (two_slash == std::string::npos)
begin_of_path = uri.find('/');
else
begin_of_path = uri.find('/', two_slash + 2);
const String path_from_uri = uri.substr(begin_of_path);
const String uri_without_path = uri.substr(0, begin_of_path);
@ -281,6 +289,9 @@ Pipe StorageHDFS::read(
auto sources_info = std::make_shared<HDFSSource::SourcesInfo>();
sources_info->uris = LSWithRegexpMatching("/", fs, path_from_uri);
if (sources_info->uris.empty())
LOG_WARNING(log, "No file in HDFS matches the path: {}", uri);
for (const auto & column : column_names)
{
if (column == "_path")

2
src/Storages/HDFS/StorageHDFS.h
View File

@ -42,7 +42,7 @@ protected:
const String & compression_method_);
private:
String uri;
const String uri;
String format_name;
String compression_method;