mirror of
https://github.com/ClickHouse/ClickHouse.git
synced 2024-11-28 10:31:57 +00:00
0fd1fc9cdc
Requires a way of deallocating keys in memory Arena (a new type of arena, actually), MySQLDictionarySource not yet capable of requesting complex keys.
146 lines
4.4 KiB
C++
146 lines
4.4 KiB
C++
#pragma once
|
|
|
|
#include <DB/Dictionaries/IDictionarySource.h>
|
|
#include <DB/Dictionaries/MongoDBBlockInputStream.h>
|
|
#include <Poco/Util/AbstractConfiguration.h>
|
|
#include <mongo/client/dbclient.h>
|
|
#include <ext/collection_cast.hpp>
|
|
|
|
|
|
namespace DB
|
|
{
|
|
|
|
/// Allows loading dictionaries from a MongoDB collection
|
|
class MongoDBDictionarySource final : public IDictionarySource
|
|
{
|
|
MongoDBDictionarySource(
|
|
const DictionaryStructure & dict_struct, const std::string & host, const std::string & port,
|
|
const std::string & user, const std::string & password,
|
|
const std::string & db, const std::string & collection,
|
|
const Block & sample_block, Context & context)
|
|
: dict_struct{dict_struct}, host{host}, port{port}, user{user}, password{password},
|
|
db{db}, collection{collection}, sample_block{sample_block}, context(context),
|
|
connection{true}
|
|
{
|
|
init();
|
|
|
|
connection.connect(host + ':' + port);
|
|
|
|
/// @todo: should connection.auth be called after or before .connect ?
|
|
if (!user.empty())
|
|
{
|
|
std::string error;
|
|
if (!connection.auth(db, user, password, error))
|
|
throw DB::Exception{
|
|
"Could not authenticate to a MongoDB database " + db + " with provided credentials: " + error,
|
|
ErrorCodes::WRONG_PASSWORD
|
|
};
|
|
}
|
|
|
|
/// compose BSONObj containing all requested fields
|
|
mongo::BSONObjBuilder builder;
|
|
builder << "_id" << 0;
|
|
|
|
for (const auto & column : sample_block.getColumns())
|
|
builder << column.name << 1;
|
|
|
|
fields_to_query = builder.obj();
|
|
}
|
|
|
|
/// mongo-cxx driver requires global initialization before using any functionality
|
|
static void init()
|
|
{
|
|
static const auto mongo_init_status = mongo::client::initialize();
|
|
|
|
if (!mongo_init_status.isOK())
|
|
throw DB::Exception{
|
|
"mongo::client::initialize() failed: " + mongo_init_status.toString(),
|
|
ErrorCodes::MONGODB_INIT_FAILED
|
|
};
|
|
|
|
LOG_TRACE(&Logger::get("MongoDBDictionarySource"), "mongo::client::initialize() ok");
|
|
}
|
|
|
|
public:
|
|
MongoDBDictionarySource(
|
|
const DictionaryStructure & dict_struct, const Poco::Util::AbstractConfiguration & config,
|
|
const std::string & config_prefix, Block & sample_block, Context & context)
|
|
: MongoDBDictionarySource{
|
|
dict_struct,
|
|
config.getString(config_prefix + ".host"),
|
|
config.getString(config_prefix + ".port"),
|
|
config.getString(config_prefix + ".user", ""),
|
|
config.getString(config_prefix + ".password", ""),
|
|
config.getString(config_prefix + ".db", ""),
|
|
config.getString(config_prefix + ".collection"),
|
|
sample_block, context
|
|
}
|
|
{
|
|
}
|
|
|
|
MongoDBDictionarySource(const MongoDBDictionarySource & other)
|
|
: MongoDBDictionarySource{
|
|
other.dict_struct, other.host, other.port, other.user, other.password,
|
|
other.db, other.collection, other.sample_block, other.context
|
|
}
|
|
{
|
|
}
|
|
|
|
BlockInputStreamPtr loadAll() override
|
|
{
|
|
return new MongoDBBlockInputStream{
|
|
connection.query(db + '.' + collection, {}, 0, 0, &fields_to_query),
|
|
sample_block, 8192
|
|
};
|
|
}
|
|
|
|
bool supportsSelectiveLoad() const override { return true; }
|
|
|
|
BlockInputStreamPtr loadIds(const std::vector<std::uint64_t> & ids) override
|
|
{
|
|
if (dict_struct.key)
|
|
throw Exception{"Complex key not supported", ErrorCodes::UNSUPPORTED_METHOD};
|
|
|
|
/// mongo::BSONObj has shitty design and does not use fixed width integral types
|
|
const auto ids_enumeration = BSON(
|
|
dict_struct.id->name << BSON("$in" << ext::collection_cast<std::vector<long long int>>(ids)));
|
|
|
|
return new MongoDBBlockInputStream{
|
|
connection.query(db + '.' + collection, ids_enumeration, 0, 0, &fields_to_query),
|
|
sample_block, 8192
|
|
};
|
|
}
|
|
|
|
BlockInputStreamPtr loadKeys(
|
|
const ConstColumnPlainPtrs & key_columns, const std::vector<std::size_t> & requested_rows) override
|
|
{
|
|
throw Exception{"Method unsupported", ErrorCodes::NOT_IMPLEMENTED};
|
|
}
|
|
|
|
/// @todo: for MongoDB, modification date can somehow be determined from the `_id` object field
|
|
bool isModified() const override { return false; }
|
|
|
|
DictionarySourcePtr clone() const override { return std::make_unique<MongoDBDictionarySource>(*this); }
|
|
|
|
std::string toString() const override
|
|
{
|
|
return "MongoDB: " + db + '.' + collection + ',' + (user.empty() ? " " : " " + user + '@') + host + ':' + port;
|
|
}
|
|
|
|
private:
|
|
const DictionaryStructure dict_struct;
|
|
const std::string host;
|
|
const std::string port;
|
|
const std::string user;
|
|
const std::string password;
|
|
const std::string db;
|
|
const std::string collection;
|
|
Block sample_block;
|
|
Context & context;
|
|
|
|
mongo::DBClientConnection connection;
|
|
mongo::BSONObj fields_to_query;
|
|
};
|
|
|
|
}
|