ClickHouse/dbms/include/DB/DataStreams/CreatingSetsBlockInputStream.h

73 lines
1.9 KiB
C++

#pragma once
#include <DB/DataStreams/IProfilingBlockInputStream.h>
#include <DB/Interpreters/ExpressionAnalyzer.h>
#include <DB/Interpreters/Set.h>
#include <DB/Interpreters/Join.h>
namespace DB
{
/** Отдает без изменений данные из потока блоков, но перед чтением первого блока инициализирует все переданные множества.
*/
class CreatingSetsBlockInputStream : public IProfilingBlockInputStream
{
public:
CreatingSetsBlockInputStream(
BlockInputStreamPtr input,
SubqueriesForSets & subqueries_for_sets_,
const Limits & limits)
: subqueries_for_sets(subqueries_for_sets_),
max_rows_to_transfer(limits.max_rows_to_transfer),
max_bytes_to_transfer(limits.max_bytes_to_transfer),
transfer_overflow_mode(limits.transfer_overflow_mode)
{
for (auto & elem : subqueries_for_sets)
children.push_back(elem.second.source);
children.push_back(input);
}
String getName() const { return "CreatingSetsBlockInputStream"; }
String getID() const
{
std::stringstream res;
res << "CreatingSets(";
Strings children_ids(children.size());
for (size_t i = 0; i < children.size(); ++i)
children_ids[i] = children[i]->getID();
/// Будем считать, что порядок создания множеств не имеет значения.
std::sort(children_ids.begin(), children_ids.end() - 1);
for (size_t i = 0; i < children_ids.size(); ++i)
res << (i == 0 ? "" : ", ") << children_ids[i];
res << ")";
return res.str();
}
protected:
Block readImpl();
private:
SubqueriesForSets subqueries_for_sets;
bool created = false;
size_t max_rows_to_transfer;
size_t max_bytes_to_transfer;
OverflowMode transfer_overflow_mode;
size_t rows_to_transfer = 0;
size_t bytes_to_transfer = 0;
Logger * log = &Logger::get("CreatingSetsBlockInputStream");
void create(SubqueryForSet & subquery);
};
}