ClickHouse/dbms/src/Dictionaries/HTTPDictionarySource.cpp

132 lines
4.2 KiB
C++
Raw Normal View History

2016-11-15 19:51:06 +00:00
#include <DB/Dictionaries/HTTPDictionarySource.h>
2016-11-25 00:16:20 +00:00
#include <Poco/Net/HTTPRequest.h>
2016-11-15 19:51:06 +00:00
#include <DB/Interpreters/Context.h>
2016-11-19 00:07:58 +00:00
#include <DB/Dictionaries/OwningBlockInputStream.h>
#include <DB/IO/ReadWriteBufferFromHTTP.h>
2016-11-15 19:51:06 +00:00
2016-11-22 15:03:54 +00:00
#include <DB/DataStreams/IBlockOutputStream.h>
#include <DB/DataTypes/DataTypesNumberFixed.h>
#include <DB/IO/WriteBufferFromOStream.h>
2016-11-15 19:51:06 +00:00
namespace DB
{
2016-11-22 15:03:54 +00:00
HTTPDictionarySource::HTTPDictionarySource(const DictionaryStructure & dict_struct_,
const Poco::Util::AbstractConfiguration & config, const std::string & config_prefix,
Block & sample_block, const Context & context) :
2016-11-18 01:48:13 +00:00
dict_struct{dict_struct_},
2016-11-24 01:01:11 +00:00
url{config.getString(config_prefix + ".url", "")},
2016-11-15 23:55:45 +00:00
format{config.getString(config_prefix + ".format")},
2016-11-15 22:05:49 +00:00
sample_block{sample_block},
context(context)
2016-11-15 19:51:06 +00:00
{
}
2016-11-15 22:05:49 +00:00
HTTPDictionarySource::HTTPDictionarySource(const HTTPDictionarySource & other) :
2016-11-18 01:48:13 +00:00
dict_struct{other.dict_struct},
2016-11-24 01:01:11 +00:00
url{other.url},
2016-11-18 01:48:13 +00:00
sample_block{other.sample_block},
context(other.context)
2016-11-15 19:51:06 +00:00
{
}
BlockInputStreamPtr HTTPDictionarySource::loadAll()
{
2016-11-19 00:07:58 +00:00
LOG_TRACE(log, "loadAll " + toString());
2016-11-24 01:01:11 +00:00
Poco::URI uri(url);
2016-11-24 19:57:24 +00:00
auto in_ptr = std::make_unique<ReadWriteBufferFromHTTP>(uri, Poco::Net::HTTPRequest::HTTP_GET);
2016-11-18 01:48:13 +00:00
auto stream = context.getInputFormat(format, *in_ptr, sample_block, max_block_size);
2016-11-24 19:57:24 +00:00
return std::make_shared<OwningBlockInputStream<ReadWriteBufferFromHTTP>>(stream, std::move(in_ptr));
2016-11-15 19:51:06 +00:00
}
BlockInputStreamPtr HTTPDictionarySource::loadIds(const std::vector<UInt64> & ids)
{
2016-11-25 20:37:06 +00:00
LOG_TRACE(log, "loadIds " + toString() + " ids=" + std::to_string(ids.size()));
2016-11-19 00:07:58 +00:00
2016-11-25 00:16:20 +00:00
ReadWriteBufferFromHTTP::OutStreamCallback out_stream_callback = [&](std::ostream & out_stream) {
2016-11-22 15:03:54 +00:00
// copypaste from ExecutableDictionarySource.cpp, todo: make func
ColumnWithTypeAndName column;
column.type = std::make_shared<DataTypeUInt64>();
column.column = column.type->createColumn();
for (auto & id : ids) {
column.column->insert(id); //CHECKME maybe faster?
}
Block block;
block.insert(std::move(column));
WriteBufferFromOStream out_buffer(out_stream);
auto stream_out = context.getOutputFormat(format, out_buffer, sample_block);
2016-11-25 20:37:06 +00:00
stream_out->writePrefix();
2016-11-22 15:03:54 +00:00
stream_out->write(block);
2016-11-25 20:37:06 +00:00
stream_out->writeSuffix();
stream_out->flush();
2016-11-22 15:03:54 +00:00
};
2016-11-24 19:57:24 +00:00
Poco::URI uri(url);
2016-11-25 00:16:20 +00:00
auto in_ptr = std::make_unique<ReadWriteBufferFromHTTP>(uri, Poco::Net::HTTPRequest::HTTP_POST, out_stream_callback);
2016-11-19 00:07:58 +00:00
auto stream = context.getInputFormat(format, *in_ptr, sample_block, max_block_size);
return std::make_shared<OwningBlockInputStream<ReadWriteBufferFromHTTP>>(stream, std::move(in_ptr));
2016-11-15 19:51:06 +00:00
}
BlockInputStreamPtr HTTPDictionarySource::loadKeys(
const ConstColumnPlainPtrs & key_columns, const std::vector<std::size_t> & requested_rows)
{
2016-11-25 20:37:06 +00:00
LOG_TRACE(log, "loadKeys " + toString() + " rows=" + std::to_string(requested_rows.size()));
2016-11-22 15:03:54 +00:00
2016-11-25 00:16:20 +00:00
ReadWriteBufferFromHTTP::OutStreamCallback out_stream_callback = [&](std::ostream & out_stream) {
// copypaste from ExecutableDictionarySource.cpp, todo: make func
2016-11-22 15:03:54 +00:00
Block block;
const auto keys_size = key_columns.size();
for (const auto i : ext::range(0, keys_size))
{
const auto & key_description = (*dict_struct.key)[i];
const auto & key = key_columns[i];
ColumnWithTypeAndName column;
column.type = key_description.type;
column.column = key->clone(); // CHECKME !!
block.insert(std::move(column));
}
WriteBufferFromOStream out_buffer(out_stream);
auto stream_out = context.getOutputFormat(format, out_buffer, sample_block);
2016-11-25 20:37:06 +00:00
stream_out->writePrefix();
2016-11-22 15:03:54 +00:00
stream_out->write(block);
2016-11-25 20:37:06 +00:00
stream_out->writeSuffix();
stream_out->flush();
2016-11-22 15:03:54 +00:00
};
2016-11-24 19:57:24 +00:00
Poco::URI uri(url);
2016-11-25 00:16:20 +00:00
auto in_ptr = std::make_unique<ReadWriteBufferFromHTTP>(uri, Poco::Net::HTTPRequest::HTTP_POST, out_stream_callback);
2016-11-22 15:03:54 +00:00
auto stream = context.getInputFormat(format, *in_ptr, sample_block, max_block_size);
return std::make_shared<OwningBlockInputStream<ReadWriteBufferFromHTTP>>(stream, std::move(in_ptr));
2016-11-15 19:51:06 +00:00
}
bool HTTPDictionarySource::isModified() const
{
2016-11-18 01:48:13 +00:00
return true;
2016-11-15 19:51:06 +00:00
}
bool HTTPDictionarySource::supportsSelectiveLoad() const
{
2016-11-23 22:44:53 +00:00
return true;
2016-11-15 19:51:06 +00:00
}
DictionarySourcePtr HTTPDictionarySource::clone() const
{
return std::make_unique<HTTPDictionarySource>(*this);
}
std::string HTTPDictionarySource::toString() const
{
2016-11-24 01:01:11 +00:00
Poco::URI uri(url);
return uri.toString();
2016-11-15 19:51:06 +00:00
}
}