2017-01-21 04:24:28 +00:00
|
|
|
#include <common/logger_useful.h>
|
|
|
|
|
2017-04-01 09:19:00 +00:00
|
|
|
#include <Dictionaries/MySQLDictionarySource.h>
|
|
|
|
#include <Dictionaries/MySQLBlockInputStream.h>
|
2016-05-29 01:12:10 +00:00
|
|
|
|
2016-06-05 15:21:35 +00:00
|
|
|
|
|
|
|
namespace DB
|
|
|
|
{
|
|
|
|
|
|
|
|
|
2016-12-08 02:49:04 +00:00
|
|
|
static const size_t max_block_size = 8192;
|
2016-06-05 15:21:35 +00:00
|
|
|
|
|
|
|
|
|
|
|
MySQLDictionarySource::MySQLDictionarySource(const DictionaryStructure & dict_struct_,
|
2017-04-01 07:20:54 +00:00
|
|
|
const Poco::Util::AbstractConfiguration & config, const std::string & config_prefix,
|
|
|
|
const Block & sample_block)
|
|
|
|
: log(&Logger::get("MySQLDictionarySource")),
|
|
|
|
dict_struct{dict_struct_},
|
|
|
|
db{config.getString(config_prefix + ".db", "")},
|
|
|
|
table{config.getString(config_prefix + ".table")},
|
|
|
|
where{config.getString(config_prefix + ".where", "")},
|
|
|
|
dont_check_update_time{config.getBool(config_prefix + ".dont_check_update_time", false)},
|
|
|
|
sample_block{sample_block},
|
|
|
|
pool{config, config_prefix},
|
|
|
|
query_builder{dict_struct, db, table, where, ExternalQueryBuilder::Backticks},
|
|
|
|
load_all_query{query_builder.composeLoadAllQuery()}
|
2016-06-05 15:21:35 +00:00
|
|
|
{
|
|
|
|
}
|
|
|
|
|
|
|
|
/// copy-constructor is provided in order to support cloneability
|
|
|
|
MySQLDictionarySource::MySQLDictionarySource(const MySQLDictionarySource & other)
|
2017-04-01 07:20:54 +00:00
|
|
|
: log(&Logger::get("MySQLDictionarySource")),
|
|
|
|
dict_struct{other.dict_struct},
|
|
|
|
db{other.db},
|
|
|
|
table{other.table},
|
|
|
|
where{other.where},
|
|
|
|
dont_check_update_time{other.dont_check_update_time},
|
|
|
|
sample_block{other.sample_block},
|
|
|
|
pool{other.pool},
|
|
|
|
query_builder{dict_struct, db, table, where, ExternalQueryBuilder::Backticks},
|
|
|
|
load_all_query{other.load_all_query}, last_modification{other.last_modification}
|
2016-06-05 15:21:35 +00:00
|
|
|
{
|
|
|
|
}
|
|
|
|
|
|
|
|
BlockInputStreamPtr MySQLDictionarySource::loadAll()
|
|
|
|
{
|
2017-04-01 07:20:54 +00:00
|
|
|
last_modification = getLastModification();
|
2016-06-05 15:21:35 +00:00
|
|
|
|
2017-04-01 07:20:54 +00:00
|
|
|
LOG_TRACE(log, load_all_query);
|
|
|
|
return std::make_shared<MySQLBlockInputStream>(pool.Get(), load_all_query, sample_block, max_block_size);
|
2016-06-05 15:21:35 +00:00
|
|
|
}
|
|
|
|
|
2016-10-26 22:27:38 +00:00
|
|
|
BlockInputStreamPtr MySQLDictionarySource::loadIds(const std::vector<UInt64> & ids)
|
2016-06-05 15:21:35 +00:00
|
|
|
{
|
2017-04-01 07:20:54 +00:00
|
|
|
/// We do not log in here and do not update the modification time, as the request can be large, and often called.
|
2016-06-05 15:21:35 +00:00
|
|
|
|
2017-04-01 07:20:54 +00:00
|
|
|
const auto query = query_builder.composeLoadIdsQuery(ids);
|
|
|
|
return std::make_shared<MySQLBlockInputStream>(pool.Get(), query, sample_block, max_block_size);
|
2016-06-05 15:21:35 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
BlockInputStreamPtr MySQLDictionarySource::loadKeys(
|
2017-04-01 07:20:54 +00:00
|
|
|
const ConstColumnPlainPtrs & key_columns, const std::vector<std::size_t> & requested_rows)
|
2016-06-05 15:21:35 +00:00
|
|
|
{
|
2017-04-01 07:20:54 +00:00
|
|
|
/// We do not log in here and do not update the modification time, as the request can be large, and often called.
|
2016-06-05 15:21:35 +00:00
|
|
|
|
2017-04-01 07:20:54 +00:00
|
|
|
const auto query = query_builder.composeLoadKeysQuery(key_columns, requested_rows, ExternalQueryBuilder::AND_OR_CHAIN);
|
|
|
|
return std::make_shared<MySQLBlockInputStream>(pool.Get(), query, sample_block, max_block_size);
|
2016-06-05 15:21:35 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
bool MySQLDictionarySource::isModified() const
|
|
|
|
{
|
2017-04-01 07:20:54 +00:00
|
|
|
if (dont_check_update_time)
|
|
|
|
return true;
|
2016-06-05 15:21:35 +00:00
|
|
|
|
2017-04-01 07:20:54 +00:00
|
|
|
return getLastModification() > last_modification;
|
2016-06-05 15:21:35 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
bool MySQLDictionarySource::supportsSelectiveLoad() const
|
|
|
|
{
|
2017-04-01 07:20:54 +00:00
|
|
|
return true;
|
2016-06-05 15:21:35 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
DictionarySourcePtr MySQLDictionarySource::clone() const
|
|
|
|
{
|
2017-04-01 07:20:54 +00:00
|
|
|
return std::make_unique<MySQLDictionarySource>(*this);
|
2016-06-05 15:21:35 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
std::string MySQLDictionarySource::toString() const
|
|
|
|
{
|
2017-04-01 07:20:54 +00:00
|
|
|
return "MySQL: " + db + '.' + table + (where.empty() ? "" : ", where: " + where);
|
2016-06-05 15:21:35 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
std::string MySQLDictionarySource::quoteForLike(const std::string s)
|
|
|
|
{
|
2017-04-01 07:20:54 +00:00
|
|
|
std::string tmp;
|
|
|
|
tmp.reserve(s.size());
|
|
|
|
|
|
|
|
for (auto c : s)
|
|
|
|
{
|
|
|
|
if (c == '%' || c == '_' || c == '\\')
|
|
|
|
tmp.push_back('\\');
|
|
|
|
tmp.push_back(c);
|
|
|
|
}
|
|
|
|
|
|
|
|
std::string res;
|
|
|
|
{
|
|
|
|
WriteBufferFromString out(res);
|
|
|
|
writeQuoted(tmp, out);
|
|
|
|
}
|
|
|
|
return res;
|
2016-06-05 15:21:35 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
LocalDateTime MySQLDictionarySource::getLastModification() const
|
|
|
|
{
|
2017-04-01 07:20:54 +00:00
|
|
|
LocalDateTime update_time{std::time(nullptr)};
|
|
|
|
|
|
|
|
if (dont_check_update_time)
|
|
|
|
return update_time;
|
|
|
|
|
|
|
|
try
|
|
|
|
{
|
|
|
|
auto connection = pool.Get();
|
|
|
|
auto query = connection->query("SHOW TABLE STATUS LIKE " + quoteForLike(table));
|
|
|
|
|
|
|
|
LOG_TRACE(log, query.str());
|
|
|
|
|
|
|
|
auto result = query.use();
|
|
|
|
|
|
|
|
size_t fetched_rows = 0;
|
|
|
|
if (auto row = result.fetch())
|
|
|
|
{
|
|
|
|
++fetched_rows;
|
|
|
|
const auto UPDATE_TIME_IDX = 12;
|
|
|
|
const auto & update_time_value = row[UPDATE_TIME_IDX];
|
|
|
|
|
|
|
|
if (!update_time_value.isNull())
|
|
|
|
{
|
|
|
|
update_time = update_time_value.getDateTime();
|
|
|
|
LOG_TRACE(log, "Got update time: " << update_time);
|
|
|
|
}
|
|
|
|
|
|
|
|
/// fetch remaining rows to avoid "commands out of sync" error
|
|
|
|
while (result.fetch())
|
|
|
|
++fetched_rows;
|
|
|
|
}
|
|
|
|
|
|
|
|
if (0 == fetched_rows)
|
|
|
|
LOG_ERROR(log, "Cannot find table in SHOW TABLE STATUS result.");
|
|
|
|
|
|
|
|
if (fetched_rows > 1)
|
|
|
|
LOG_ERROR(log, "Found more than one table in SHOW TABLE STATUS result.");
|
|
|
|
}
|
|
|
|
catch (...)
|
|
|
|
{
|
|
|
|
tryLogCurrentException("MySQLDictionarySource");
|
|
|
|
}
|
|
|
|
|
|
|
|
/// we suppose failure to get modification time is not an error, therefore return current time
|
|
|
|
return update_time;
|
2016-06-05 15:21:35 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
}
|