2011-08-28 05:13:24 +00:00
|
|
|
|
#include <DB/DataStreams/ExpressionBlockInputStream.h>
|
|
|
|
|
#include <DB/DataStreams/ProjectionBlockInputStream.h>
|
|
|
|
|
#include <DB/DataStreams/FilterBlockInputStream.h>
|
|
|
|
|
#include <DB/DataStreams/LimitBlockInputStream.h>
|
2011-09-04 05:14:52 +00:00
|
|
|
|
#include <DB/DataStreams/PartialSortingBlockInputStream.h>
|
|
|
|
|
#include <DB/DataStreams/MergeSortingBlockInputStream.h>
|
2011-09-25 03:37:09 +00:00
|
|
|
|
#include <DB/DataStreams/AggregatingBlockInputStream.h>
|
|
|
|
|
#include <DB/DataStreams/FinalizingAggregatedBlockInputStream.h>
|
2011-10-30 05:19:41 +00:00
|
|
|
|
#include <DB/DataStreams/FormatFactory.h>
|
|
|
|
|
#include <DB/DataStreams/copyData.h>
|
2011-08-28 05:13:24 +00:00
|
|
|
|
|
|
|
|
|
#include <DB/Parsers/ASTSelectQuery.h>
|
|
|
|
|
#include <DB/Parsers/ASTIdentifier.h>
|
2011-09-04 05:14:52 +00:00
|
|
|
|
#include <DB/Parsers/ASTFunction.h>
|
2011-08-28 08:02:11 +00:00
|
|
|
|
#include <DB/Parsers/ASTLiteral.h>
|
2011-09-04 05:14:52 +00:00
|
|
|
|
#include <DB/Parsers/ASTOrderByElement.h>
|
2011-08-28 05:13:24 +00:00
|
|
|
|
|
|
|
|
|
#include <DB/Interpreters/Expression.h>
|
|
|
|
|
#include <DB/Interpreters/InterpreterSelectQuery.h>
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
namespace DB
|
|
|
|
|
{
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
InterpreterSelectQuery::InterpreterSelectQuery(ASTPtr query_ptr_, Context & context_, size_t max_block_size_)
|
|
|
|
|
: query_ptr(query_ptr_), context(context_), max_block_size(max_block_size_)
|
|
|
|
|
{
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
StoragePtr InterpreterSelectQuery::getTable()
|
|
|
|
|
{
|
|
|
|
|
ASTSelectQuery & query = dynamic_cast<ASTSelectQuery &>(*query_ptr);
|
|
|
|
|
|
|
|
|
|
/// Из какой таблицы читать данные. JOIN-ы не поддерживаются.
|
|
|
|
|
|
|
|
|
|
String database_name;
|
|
|
|
|
String table_name;
|
|
|
|
|
|
|
|
|
|
/** Если таблица не указана - используем таблицу system.one.
|
|
|
|
|
* Если база данных не указана - используем текущую базу данных.
|
|
|
|
|
*/
|
|
|
|
|
if (!query.table)
|
|
|
|
|
{
|
|
|
|
|
database_name = "system";
|
|
|
|
|
table_name = "one";
|
|
|
|
|
}
|
|
|
|
|
else if (!query.database)
|
|
|
|
|
database_name = context.current_database;
|
|
|
|
|
|
|
|
|
|
if (query.database)
|
|
|
|
|
database_name = dynamic_cast<ASTIdentifier &>(*query.database).name;
|
|
|
|
|
if (query.table)
|
|
|
|
|
table_name = dynamic_cast<ASTIdentifier &>(*query.table).name;
|
|
|
|
|
|
|
|
|
|
if (context.databases->end() == context.databases->find(database_name)
|
|
|
|
|
|| (*context.databases)[database_name].end() == (*context.databases)[database_name].find(table_name))
|
2011-08-28 08:02:11 +00:00
|
|
|
|
throw Exception("Unknown table '" + table_name + "' in database '" + database_name + "'", ErrorCodes::UNKNOWN_TABLE);
|
2011-08-28 05:13:24 +00:00
|
|
|
|
|
|
|
|
|
return (*context.databases)[database_name][table_name];
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
2011-11-06 02:29:13 +00:00
|
|
|
|
void InterpreterSelectQuery::setColumns()
|
|
|
|
|
{
|
|
|
|
|
ASTSelectQuery & query = dynamic_cast<ASTSelectQuery &>(*query_ptr);
|
|
|
|
|
|
2011-11-06 22:00:39 +00:00
|
|
|
|
context.columns = !query.table || !dynamic_cast<ASTSelectQuery *>(&*query.table)
|
2011-11-06 02:29:13 +00:00
|
|
|
|
? getTable()->getColumnsList()
|
|
|
|
|
: InterpreterSelectQuery(query.table, context, max_block_size).getSampleBlock().getColumnsList();
|
|
|
|
|
|
|
|
|
|
if (context.columns.empty())
|
|
|
|
|
throw Exception("There is no available columns", ErrorCodes::THERE_IS_NO_COLUMN);
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
2011-08-28 05:13:24 +00:00
|
|
|
|
DataTypes InterpreterSelectQuery::getReturnTypes()
|
|
|
|
|
{
|
2011-11-06 02:29:13 +00:00
|
|
|
|
setColumns();
|
2011-08-28 05:13:24 +00:00
|
|
|
|
Expression expression(dynamic_cast<ASTSelectQuery &>(*query_ptr).select_expression_list, context);
|
|
|
|
|
return expression.getReturnTypes();
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
2011-10-30 05:19:41 +00:00
|
|
|
|
Block InterpreterSelectQuery::getSampleBlock()
|
|
|
|
|
{
|
2011-11-06 02:29:13 +00:00
|
|
|
|
setColumns();
|
2011-10-30 05:19:41 +00:00
|
|
|
|
Expression expression(dynamic_cast<ASTSelectQuery &>(*query_ptr).select_expression_list, context);
|
|
|
|
|
return expression.getSampleBlock();
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
2011-08-28 05:13:24 +00:00
|
|
|
|
BlockInputStreamPtr InterpreterSelectQuery::execute()
|
|
|
|
|
{
|
|
|
|
|
ASTSelectQuery & query = dynamic_cast<ASTSelectQuery &>(*query_ptr);
|
|
|
|
|
|
2011-11-06 02:29:13 +00:00
|
|
|
|
/// Таблица, откуда читать данные, если не подзапрос.
|
|
|
|
|
StoragePtr table;
|
|
|
|
|
/// Интерпретатор подзапроса, если подзапрос
|
|
|
|
|
SharedPtr<InterpreterSelectQuery> interpreter_subquery;
|
2011-08-28 05:13:24 +00:00
|
|
|
|
|
2011-11-06 02:29:13 +00:00
|
|
|
|
/// Добавляем в контекст список доступных столбцов.
|
|
|
|
|
setColumns();
|
|
|
|
|
|
2011-11-06 22:00:39 +00:00
|
|
|
|
if (!query.table || !dynamic_cast<ASTSelectQuery *>(&*query.table))
|
2011-11-06 02:29:13 +00:00
|
|
|
|
table = getTable();
|
|
|
|
|
else
|
|
|
|
|
interpreter_subquery = new InterpreterSelectQuery(query.table, context, max_block_size);
|
|
|
|
|
|
2011-11-06 04:21:09 +00:00
|
|
|
|
/// Объект, с помощью которого анализируется запрос.
|
2011-08-28 05:13:24 +00:00
|
|
|
|
Poco::SharedPtr<Expression> expression = new Expression(query_ptr, context);
|
2011-11-06 02:29:13 +00:00
|
|
|
|
/// Список столбцов, которых нужно прочитать, чтобы выполнить запрос.
|
2011-08-28 08:02:11 +00:00
|
|
|
|
Names required_columns = expression->getRequiredColumns();
|
|
|
|
|
|
|
|
|
|
/// Если не указан ни один столбец из таблицы, то будем читать первый попавшийся (чтобы хотя бы знать число строк).
|
|
|
|
|
if (required_columns.empty())
|
2011-11-06 02:29:13 +00:00
|
|
|
|
required_columns.push_back(context.columns.front().first);
|
|
|
|
|
|
|
|
|
|
/// Нужно ли агрегировать.
|
|
|
|
|
bool need_aggregate = expression->hasAggregates() || query.group_expression_list;
|
2011-08-28 08:02:11 +00:00
|
|
|
|
|
2011-09-04 05:14:52 +00:00
|
|
|
|
size_t limit_length = 0;
|
|
|
|
|
size_t limit_offset = 0;
|
|
|
|
|
if (query.limit_length)
|
|
|
|
|
{
|
|
|
|
|
limit_length = boost::get<UInt64>(dynamic_cast<ASTLiteral &>(*query.limit_length).value);
|
|
|
|
|
if (query.limit_offset)
|
|
|
|
|
limit_offset = boost::get<UInt64>(dynamic_cast<ASTLiteral &>(*query.limit_offset).value);
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
/** Оптимизация - если не указаны WHERE, GROUP, HAVING, ORDER, но указан LIMIT, и limit + offset < max_block_size,
|
|
|
|
|
* то в качестве размера блока будем использовать limit + offset (чтобы не читать из таблицы больше, чем запрошено).
|
|
|
|
|
*/
|
|
|
|
|
size_t block_size = max_block_size;
|
|
|
|
|
if (!query.where_expression && !query.group_expression_list && !query.having_expression && !query.order_expression_list
|
2011-09-25 06:42:59 +00:00
|
|
|
|
&& query.limit_length && !need_aggregate && limit_length + limit_offset < block_size)
|
2011-09-04 05:14:52 +00:00
|
|
|
|
{
|
|
|
|
|
block_size = limit_length + limit_offset;
|
|
|
|
|
}
|
|
|
|
|
|
2011-11-06 02:29:13 +00:00
|
|
|
|
/// Поток данных.
|
|
|
|
|
BlockInputStreamPtr stream;
|
|
|
|
|
|
|
|
|
|
/// Инициализируем изначальный поток данных, на который накладываются преобразования запроса. Таблица или подзапрос?
|
2011-11-06 22:00:39 +00:00
|
|
|
|
if (!query.table || !dynamic_cast<ASTSelectQuery *>(&*query.table))
|
2011-11-06 02:29:13 +00:00
|
|
|
|
stream = table->read(required_columns, query_ptr, block_size);
|
|
|
|
|
else
|
|
|
|
|
stream = interpreter_subquery->execute();
|
2011-08-28 05:13:24 +00:00
|
|
|
|
|
2011-09-25 03:37:09 +00:00
|
|
|
|
bool is_first_expression = true;
|
|
|
|
|
|
2011-08-28 05:13:24 +00:00
|
|
|
|
/// Если есть условие WHERE - сначала выполним часть выражения, необходимую для его вычисления
|
|
|
|
|
if (query.where_expression)
|
|
|
|
|
{
|
|
|
|
|
setPartID(query.where_expression, PART_WHERE);
|
2011-09-25 03:37:09 +00:00
|
|
|
|
stream = new ExpressionBlockInputStream(stream, expression, is_first_expression, PART_WHERE);
|
2011-08-28 05:13:24 +00:00
|
|
|
|
stream = new FilterBlockInputStream(stream);
|
2011-11-06 22:00:39 +00:00
|
|
|
|
is_first_expression = false;
|
2011-08-28 05:13:24 +00:00
|
|
|
|
}
|
|
|
|
|
|
2011-09-25 03:37:09 +00:00
|
|
|
|
/// Если есть GROUP BY - сначала выполним часть выражения, необходимую для его вычисления
|
2011-09-25 06:42:59 +00:00
|
|
|
|
if (need_aggregate)
|
2011-09-25 03:37:09 +00:00
|
|
|
|
{
|
|
|
|
|
expression->markBeforeAndAfterAggregation(PART_BEFORE_AGGREGATING, PART_AFTER_AGGREGATING);
|
|
|
|
|
|
|
|
|
|
if (query.group_expression_list)
|
|
|
|
|
setPartID(query.group_expression_list, PART_GROUP);
|
|
|
|
|
|
|
|
|
|
stream = new ExpressionBlockInputStream(stream, expression, is_first_expression, PART_GROUP | PART_BEFORE_AGGREGATING);
|
|
|
|
|
stream = new AggregatingBlockInputStream(stream, expression);
|
|
|
|
|
stream = new FinalizingAggregatedBlockInputStream(stream);
|
2011-11-06 22:00:39 +00:00
|
|
|
|
is_first_expression = false;
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
/// Если есть условие HAVING - сначала выполним часть выражения, необходимую для его вычисления
|
|
|
|
|
if (query.having_expression)
|
|
|
|
|
{
|
|
|
|
|
setPartID(query.having_expression, PART_HAVING);
|
|
|
|
|
stream = new ExpressionBlockInputStream(stream, expression, is_first_expression, PART_HAVING);
|
|
|
|
|
stream = new FilterBlockInputStream(stream);
|
2011-09-25 03:37:09 +00:00
|
|
|
|
is_first_expression = false;
|
|
|
|
|
}
|
|
|
|
|
|
2011-08-28 05:13:24 +00:00
|
|
|
|
/// Выполним оставшуюся часть выражения
|
|
|
|
|
setPartID(query.select_expression_list, PART_SELECT);
|
2011-09-04 05:14:52 +00:00
|
|
|
|
if (query.order_expression_list)
|
|
|
|
|
setPartID(query.order_expression_list, PART_ORDER);
|
2011-09-25 03:37:09 +00:00
|
|
|
|
stream = new ExpressionBlockInputStream(stream, expression, is_first_expression, PART_SELECT | PART_ORDER);
|
|
|
|
|
is_first_expression = false;
|
2011-09-05 00:32:22 +00:00
|
|
|
|
|
2011-11-06 21:23:23 +00:00
|
|
|
|
/** Оставим только столбцы, нужные для SELECT и ORDER BY части.
|
|
|
|
|
* Если нет ORDER BY - то это последняя проекция, и нужно брать только столбцы из SELECT части.
|
|
|
|
|
*/
|
|
|
|
|
stream = new ProjectionBlockInputStream(stream, expression,
|
|
|
|
|
query.order_expression_list ? true : false,
|
|
|
|
|
PART_SELECT | PART_ORDER,
|
|
|
|
|
query.order_expression_list ? NULL : query.select_expression_list);
|
2011-09-04 05:14:52 +00:00
|
|
|
|
|
|
|
|
|
/// Если есть ORDER BY
|
|
|
|
|
if (query.order_expression_list)
|
|
|
|
|
{
|
|
|
|
|
SortDescription order_descr;
|
|
|
|
|
order_descr.reserve(query.order_expression_list->children.size());
|
|
|
|
|
for (ASTs::iterator it = query.order_expression_list->children.begin();
|
|
|
|
|
it != query.order_expression_list->children.end();
|
|
|
|
|
++it)
|
|
|
|
|
{
|
2011-09-26 01:50:32 +00:00
|
|
|
|
String name = (*it)->children.front()->getColumnName();
|
2011-09-04 05:14:52 +00:00
|
|
|
|
order_descr.push_back(SortColumnDescription(name, dynamic_cast<ASTOrderByElement &>(**it).direction));
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
stream = new PartialSortingBlockInputStream(stream, order_descr);
|
|
|
|
|
stream = new MergeSortingBlockInputStream(stream, order_descr);
|
2011-09-05 00:32:22 +00:00
|
|
|
|
|
|
|
|
|
/// Оставим только столбцы, нужные для SELECT части
|
2011-09-25 03:37:09 +00:00
|
|
|
|
stream = new ProjectionBlockInputStream(stream, expression, false, PART_SELECT, query.select_expression_list);
|
2011-09-04 05:14:52 +00:00
|
|
|
|
}
|
|
|
|
|
|
2011-08-28 08:02:11 +00:00
|
|
|
|
/// Если есть LIMIT
|
|
|
|
|
if (query.limit_length)
|
|
|
|
|
{
|
|
|
|
|
stream = new LimitBlockInputStream(stream, limit_length, limit_offset);
|
|
|
|
|
}
|
|
|
|
|
|
2011-08-28 05:13:24 +00:00
|
|
|
|
return stream;
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
2011-10-30 05:19:41 +00:00
|
|
|
|
BlockInputStreamPtr InterpreterSelectQuery::executeAndFormat(WriteBuffer & buf)
|
|
|
|
|
{
|
|
|
|
|
FormatFactory format_factory;
|
|
|
|
|
ASTSelectQuery & query = dynamic_cast<ASTSelectQuery &>(*query_ptr);
|
|
|
|
|
Block sample = getSampleBlock();
|
|
|
|
|
String format_name = query.format ? dynamic_cast<ASTIdentifier &>(*query.format).name : "TabSeparated";
|
|
|
|
|
|
|
|
|
|
BlockInputStreamPtr in = execute();
|
|
|
|
|
BlockOutputStreamPtr out = format_factory.getOutput(format_name, buf, sample);
|
|
|
|
|
|
|
|
|
|
copyData(*in, *out);
|
|
|
|
|
|
|
|
|
|
return in;
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
2011-08-28 05:13:24 +00:00
|
|
|
|
void InterpreterSelectQuery::setPartID(ASTPtr ast, unsigned part_id)
|
|
|
|
|
{
|
2011-09-04 05:14:52 +00:00
|
|
|
|
ast->part_id |= part_id;
|
2011-10-30 05:19:41 +00:00
|
|
|
|
|
2011-08-28 05:13:24 +00:00
|
|
|
|
for (ASTs::iterator it = ast->children.begin(); it != ast->children.end(); ++it)
|
|
|
|
|
setPartID(*it, part_id);
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
}
|