2018-11-28 11:37:12 +00:00
|
|
|
#include "ExecutableDictionarySource.h"
|
|
|
|
|
2017-05-25 20:59:14 +00:00
|
|
|
#include <future>
|
2018-12-10 15:25:45 +00:00
|
|
|
#include <thread>
|
2019-03-01 16:38:52 +00:00
|
|
|
#include <ext/scope_guard.h>
|
2017-04-01 09:19:00 +00:00
|
|
|
#include <DataStreams/IBlockOutputStream.h>
|
2018-12-10 15:25:45 +00:00
|
|
|
#include <DataStreams/OwningBlockInputStream.h>
|
|
|
|
#include <Interpreters/Context.h>
|
2019-03-01 16:38:52 +00:00
|
|
|
#include <IO/WriteHelpers.h>
|
2018-12-10 15:25:45 +00:00
|
|
|
#include <Common/ShellCommand.h>
|
2019-01-14 19:22:09 +00:00
|
|
|
#include <Common/ThreadPool.h>
|
2017-01-21 04:24:28 +00:00
|
|
|
#include <common/logger_useful.h>
|
2019-03-01 16:38:52 +00:00
|
|
|
#include <common/LocalDateTime.h>
|
2018-11-28 11:37:12 +00:00
|
|
|
#include "DictionarySourceFactory.h"
|
2018-12-10 15:25:45 +00:00
|
|
|
#include "DictionarySourceHelpers.h"
|
2018-11-28 11:37:12 +00:00
|
|
|
#include "DictionaryStructure.h"
|
2019-12-15 06:34:43 +00:00
|
|
|
#include "registerDictionaries.h"
|
2017-01-21 04:24:28 +00:00
|
|
|
|
2016-11-15 19:51:06 +00:00
|
|
|
namespace DB
|
|
|
|
{
|
2019-02-10 16:55:12 +00:00
|
|
|
static const UInt64 max_block_size = 8192;
|
2016-12-08 02:49:04 +00:00
|
|
|
|
2019-12-10 17:27:29 +00:00
|
|
|
namespace ErrorCodes
|
|
|
|
{
|
2020-02-25 18:02:41 +00:00
|
|
|
extern const int LOGICAL_ERROR;
|
2019-12-10 17:27:29 +00:00
|
|
|
extern const int DICTIONARY_ACCESS_DENIED;
|
|
|
|
}
|
2016-12-08 02:49:04 +00:00
|
|
|
|
2017-10-12 19:40:42 +00:00
|
|
|
namespace
|
|
|
|
{
|
2018-12-10 15:25:45 +00:00
|
|
|
/// Owns ShellCommand and calls wait for it.
|
|
|
|
class ShellCommandOwningBlockInputStream : public OwningBlockInputStream<ShellCommand>
|
2017-10-02 13:08:09 +00:00
|
|
|
{
|
2018-12-10 15:25:45 +00:00
|
|
|
public:
|
2019-01-07 10:40:58 +00:00
|
|
|
ShellCommandOwningBlockInputStream(const BlockInputStreamPtr & impl, std::unique_ptr<ShellCommand> own_)
|
|
|
|
: OwningBlockInputStream(std::move(impl), std::move(own_))
|
2018-12-10 15:25:45 +00:00
|
|
|
{
|
|
|
|
}
|
2017-10-02 13:08:09 +00:00
|
|
|
|
2018-12-10 15:25:45 +00:00
|
|
|
void readSuffix() override
|
|
|
|
{
|
|
|
|
OwningBlockInputStream<ShellCommand>::readSuffix();
|
|
|
|
own->wait();
|
|
|
|
}
|
|
|
|
};
|
2017-10-02 13:08:09 +00:00
|
|
|
|
2017-10-12 19:40:42 +00:00
|
|
|
}
|
|
|
|
|
2017-10-12 19:35:01 +00:00
|
|
|
|
2018-12-10 15:25:45 +00:00
|
|
|
ExecutableDictionarySource::ExecutableDictionarySource(
|
|
|
|
const DictionaryStructure & dict_struct_,
|
|
|
|
const Poco::Util::AbstractConfiguration & config,
|
|
|
|
const std::string & config_prefix,
|
2019-08-03 11:02:40 +00:00
|
|
|
Block & sample_block_,
|
|
|
|
const Context & context_)
|
2020-05-30 21:57:37 +00:00
|
|
|
: log(&Poco::Logger::get("ExecutableDictionarySource"))
|
2018-12-10 15:25:45 +00:00
|
|
|
, dict_struct{dict_struct_}
|
|
|
|
, command{config.getString(config_prefix + ".command")}
|
|
|
|
, update_field{config.getString(config_prefix + ".update_field", "")}
|
|
|
|
, format{config.getString(config_prefix + ".format")}
|
2019-08-03 11:02:40 +00:00
|
|
|
, sample_block{sample_block_}
|
|
|
|
, context(context_)
|
2016-11-15 19:51:06 +00:00
|
|
|
{
|
|
|
|
}
|
|
|
|
|
2017-01-21 04:24:28 +00:00
|
|
|
ExecutableDictionarySource::ExecutableDictionarySource(const ExecutableDictionarySource & other)
|
2020-05-30 21:57:37 +00:00
|
|
|
: log(&Poco::Logger::get("ExecutableDictionarySource"))
|
2018-12-10 15:25:45 +00:00
|
|
|
, update_time{other.update_time}
|
|
|
|
, dict_struct{other.dict_struct}
|
|
|
|
, command{other.command}
|
|
|
|
, update_field{other.update_field}
|
|
|
|
, format{other.format}
|
|
|
|
, sample_block{other.sample_block}
|
|
|
|
, context(other.context)
|
2016-11-15 19:51:06 +00:00
|
|
|
{
|
|
|
|
}
|
|
|
|
|
|
|
|
BlockInputStreamPtr ExecutableDictionarySource::loadAll()
|
|
|
|
{
|
2020-05-23 22:24:01 +00:00
|
|
|
LOG_TRACE(log, "loadAll {}", toString());
|
2017-04-01 07:20:54 +00:00
|
|
|
auto process = ShellCommand::execute(command);
|
2017-05-25 19:21:57 +00:00
|
|
|
auto input_stream = context.getInputFormat(format, process->out, sample_block, max_block_size);
|
2017-10-02 13:08:09 +00:00
|
|
|
return std::make_shared<ShellCommandOwningBlockInputStream>(input_stream, std::move(process));
|
2016-11-27 13:01:35 +00:00
|
|
|
}
|
|
|
|
|
2018-02-15 13:08:23 +00:00
|
|
|
BlockInputStreamPtr ExecutableDictionarySource::loadUpdatedAll()
|
|
|
|
{
|
2019-03-01 16:38:52 +00:00
|
|
|
time_t new_update_time = time(nullptr);
|
|
|
|
SCOPE_EXIT(update_time = new_update_time);
|
|
|
|
|
|
|
|
std::string command_with_update_field = command;
|
|
|
|
if (update_time)
|
|
|
|
command_with_update_field += " " + update_field + " " + DB::toString(LocalDateTime(update_time - 1));
|
|
|
|
|
2020-05-23 22:24:01 +00:00
|
|
|
LOG_TRACE(log, "loadUpdatedAll {}", command_with_update_field);
|
2019-03-01 16:38:52 +00:00
|
|
|
auto process = ShellCommand::execute(command_with_update_field);
|
2018-02-15 13:08:23 +00:00
|
|
|
auto input_stream = context.getInputFormat(format, process->out, sample_block, max_block_size);
|
|
|
|
return std::make_shared<ShellCommandOwningBlockInputStream>(input_stream, std::move(process));
|
|
|
|
}
|
2017-05-25 20:59:14 +00:00
|
|
|
|
2017-10-12 19:40:42 +00:00
|
|
|
namespace
|
|
|
|
{
|
2018-12-10 15:25:45 +00:00
|
|
|
/** A stream, that also runs and waits for background thread
|
2017-05-25 20:59:14 +00:00
|
|
|
* (that will feed data into pipe to be read from the other side of the pipe).
|
|
|
|
*/
|
2019-01-23 14:48:50 +00:00
|
|
|
class BlockInputStreamWithBackgroundThread final : public IBlockInputStream
|
2017-05-25 20:59:14 +00:00
|
|
|
{
|
2018-12-10 15:25:45 +00:00
|
|
|
public:
|
|
|
|
BlockInputStreamWithBackgroundThread(
|
|
|
|
const BlockInputStreamPtr & stream_, std::unique_ptr<ShellCommand> && command_, std::packaged_task<void()> && task_)
|
|
|
|
: stream{stream_}, command{std::move(command_)}, task(std::move(task_)), thread([this] {
|
|
|
|
task();
|
|
|
|
command->in.close();
|
|
|
|
})
|
|
|
|
{
|
|
|
|
children.push_back(stream);
|
|
|
|
}
|
2017-05-25 20:59:14 +00:00
|
|
|
|
2018-12-10 15:25:45 +00:00
|
|
|
~BlockInputStreamWithBackgroundThread() override
|
2017-05-25 20:59:14 +00:00
|
|
|
{
|
2018-12-10 15:25:45 +00:00
|
|
|
if (thread.joinable())
|
2017-05-25 20:59:14 +00:00
|
|
|
{
|
2018-12-10 15:25:45 +00:00
|
|
|
try
|
|
|
|
{
|
|
|
|
readSuffix();
|
|
|
|
}
|
|
|
|
catch (...)
|
|
|
|
{
|
|
|
|
tryLogCurrentException(__PRETTY_FUNCTION__);
|
|
|
|
}
|
2017-05-25 20:59:14 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2018-12-10 15:25:45 +00:00
|
|
|
Block getHeader() const override { return stream->getHeader(); }
|
2018-01-06 18:10:44 +00:00
|
|
|
|
2018-12-10 15:25:45 +00:00
|
|
|
private:
|
|
|
|
Block readImpl() override { return stream->read(); }
|
2017-05-25 20:59:14 +00:00
|
|
|
|
2018-12-10 15:25:45 +00:00
|
|
|
void readSuffix() override
|
2017-10-02 13:08:09 +00:00
|
|
|
{
|
2019-01-23 14:48:50 +00:00
|
|
|
IBlockInputStream::readSuffix();
|
2018-12-10 15:25:45 +00:00
|
|
|
if (!wait_called)
|
|
|
|
{
|
|
|
|
wait_called = true;
|
|
|
|
command->wait();
|
|
|
|
}
|
|
|
|
thread.join();
|
|
|
|
/// To rethrow an exception, if any.
|
|
|
|
task.get_future().get();
|
2017-10-02 13:08:09 +00:00
|
|
|
}
|
2017-05-25 20:59:14 +00:00
|
|
|
|
2018-12-10 15:25:45 +00:00
|
|
|
String getName() const override { return "WithBackgroundThread"; }
|
2017-05-25 20:59:14 +00:00
|
|
|
|
2018-12-10 15:25:45 +00:00
|
|
|
BlockInputStreamPtr stream;
|
|
|
|
std::unique_ptr<ShellCommand> command;
|
|
|
|
std::packaged_task<void()> task;
|
2019-01-14 19:22:09 +00:00
|
|
|
ThreadFromGlobalPool thread;
|
2018-12-10 15:25:45 +00:00
|
|
|
bool wait_called = false;
|
|
|
|
};
|
2017-05-25 20:59:14 +00:00
|
|
|
|
2017-10-12 19:40:42 +00:00
|
|
|
}
|
|
|
|
|
2017-05-25 20:59:14 +00:00
|
|
|
|
2016-11-27 13:01:35 +00:00
|
|
|
BlockInputStreamPtr ExecutableDictionarySource::loadIds(const std::vector<UInt64> & ids)
|
|
|
|
{
|
2020-05-23 22:24:01 +00:00
|
|
|
LOG_TRACE(log, "loadIds {} size = {}", toString(), ids.size());
|
2017-04-01 07:20:54 +00:00
|
|
|
auto process = ShellCommand::execute(command);
|
2017-05-25 18:24:47 +00:00
|
|
|
|
2017-05-25 19:21:57 +00:00
|
|
|
auto output_stream = context.getOutputFormat(format, process->in, sample_block);
|
|
|
|
auto input_stream = context.getInputFormat(format, process->out, sample_block, max_block_size);
|
2017-05-25 20:59:14 +00:00
|
|
|
|
|
|
|
return std::make_shared<BlockInputStreamWithBackgroundThread>(
|
2018-12-10 15:25:45 +00:00
|
|
|
input_stream, std::move(process), std::packaged_task<void()>([output_stream, &ids]() mutable { formatIDs(output_stream, ids); }));
|
2016-11-15 19:51:06 +00:00
|
|
|
}
|
|
|
|
|
2018-12-10 15:25:45 +00:00
|
|
|
BlockInputStreamPtr ExecutableDictionarySource::loadKeys(const Columns & key_columns, const std::vector<size_t> & requested_rows)
|
2016-11-15 19:51:06 +00:00
|
|
|
{
|
2020-05-23 22:24:01 +00:00
|
|
|
LOG_TRACE(log, "loadKeys {} size = {}", toString(), requested_rows.size());
|
2017-04-01 07:20:54 +00:00
|
|
|
auto process = ShellCommand::execute(command);
|
2016-11-18 01:48:13 +00:00
|
|
|
|
2017-05-25 19:21:57 +00:00
|
|
|
auto output_stream = context.getOutputFormat(format, process->in, sample_block);
|
|
|
|
auto input_stream = context.getInputFormat(format, process->out, sample_block, max_block_size);
|
2017-05-25 20:59:14 +00:00
|
|
|
|
|
|
|
return std::make_shared<BlockInputStreamWithBackgroundThread>(
|
2018-12-10 15:50:58 +00:00
|
|
|
input_stream, std::move(process), std::packaged_task<void()>([output_stream, key_columns, &requested_rows, this]() mutable
|
|
|
|
{
|
2017-05-25 20:59:14 +00:00
|
|
|
formatKeys(dict_struct, output_stream, key_columns, requested_rows);
|
|
|
|
}));
|
2016-11-15 19:51:06 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
bool ExecutableDictionarySource::isModified() const
|
|
|
|
{
|
2017-04-01 07:20:54 +00:00
|
|
|
return true;
|
2016-11-15 19:51:06 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
bool ExecutableDictionarySource::supportsSelectiveLoad() const
|
|
|
|
{
|
2017-04-01 07:20:54 +00:00
|
|
|
return true;
|
2016-11-15 19:51:06 +00:00
|
|
|
}
|
|
|
|
|
2018-01-15 12:44:39 +00:00
|
|
|
bool ExecutableDictionarySource::hasUpdateField() const
|
|
|
|
{
|
2020-03-09 03:14:24 +00:00
|
|
|
return !update_field.empty();
|
2018-01-15 12:44:39 +00:00
|
|
|
}
|
|
|
|
|
2016-11-15 19:51:06 +00:00
|
|
|
DictionarySourcePtr ExecutableDictionarySource::clone() const
|
|
|
|
{
|
2017-04-01 07:20:54 +00:00
|
|
|
return std::make_unique<ExecutableDictionarySource>(*this);
|
2016-11-15 19:51:06 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
std::string ExecutableDictionarySource::toString() const
|
|
|
|
{
|
2017-04-01 07:20:54 +00:00
|
|
|
return "Executable: " + command;
|
2016-11-15 19:51:06 +00:00
|
|
|
}
|
|
|
|
|
2018-11-28 11:37:12 +00:00
|
|
|
void registerDictionarySourceExecutable(DictionarySourceFactory & factory)
|
|
|
|
{
|
2020-03-23 02:12:31 +00:00
|
|
|
auto create_table_source = [=](const DictionaryStructure & dict_struct,
|
2018-11-28 11:37:12 +00:00
|
|
|
const Poco::Util::AbstractConfiguration & config,
|
|
|
|
const std::string & config_prefix,
|
|
|
|
Block & sample_block,
|
2019-12-10 17:27:29 +00:00
|
|
|
const Context & context,
|
2020-08-15 03:10:57 +00:00
|
|
|
const std::string & /* default_database */,
|
2019-12-10 17:27:29 +00:00
|
|
|
bool check_config) -> DictionarySourcePtr
|
2019-01-29 19:05:32 +00:00
|
|
|
{
|
2018-11-28 11:37:12 +00:00
|
|
|
if (dict_struct.has_expressions)
|
2018-12-10 15:25:45 +00:00
|
|
|
throw Exception{"Dictionary source of type `executable` does not support attribute expressions", ErrorCodes::LOGICAL_ERROR};
|
2018-11-28 11:37:12 +00:00
|
|
|
|
2019-12-10 17:27:29 +00:00
|
|
|
/// Executable dictionaries may execute arbitrary commands.
|
|
|
|
/// It's OK for dictionaries created by administrator from xml-file, but
|
|
|
|
/// maybe dangerous for dictionaries created from DDL-queries.
|
|
|
|
if (check_config)
|
|
|
|
throw Exception("Dictionaries with Executable dictionary source is not allowed", ErrorCodes::DICTIONARY_ACCESS_DENIED);
|
|
|
|
|
2020-04-03 21:32:06 +00:00
|
|
|
Context context_local_copy = copyContextAndApplySettings(config_prefix, context, config);
|
|
|
|
|
2019-12-10 17:27:29 +00:00
|
|
|
return std::make_unique<ExecutableDictionarySource>(
|
|
|
|
dict_struct, config, config_prefix + ".executable",
|
2020-04-03 21:32:06 +00:00
|
|
|
sample_block, context_local_copy);
|
2018-11-28 11:37:12 +00:00
|
|
|
};
|
2020-03-23 02:12:31 +00:00
|
|
|
factory.registerSource("executable", create_table_source);
|
2018-11-28 11:37:12 +00:00
|
|
|
}
|
|
|
|
|
2016-11-15 19:51:06 +00:00
|
|
|
}
|