2020-01-14 03:07:31 +00:00
|
|
|
#include <AggregateFunctions/AggregateFunctionFactory.h>
|
|
|
|
#include <Columns/ColumnTuple.h>
|
|
|
|
#include <Columns/ColumnsNumber.h>
|
|
|
|
#include <DataStreams/BlocksBlockInputStream.h>
|
|
|
|
#include <DataStreams/IBlockOutputStream.h>
|
|
|
|
#include <DataStreams/MaterializingBlockInputStream.h>
|
|
|
|
#include <DataStreams/NullBlockInputStream.h>
|
|
|
|
#include <DataStreams/OneBlockInputStream.h>
|
|
|
|
#include <DataStreams/SquashingBlockInputStream.h>
|
|
|
|
#include <DataStreams/copyData.h>
|
|
|
|
#include <DataTypes/DataTypeArray.h>
|
|
|
|
#include <DataTypes/DataTypeDateTime.h>
|
|
|
|
#include <DataTypes/DataTypesNumber.h>
|
|
|
|
#include <Functions/FunctionsWindow.h>
|
|
|
|
#include <Interpreters/AddDefaultDatabaseVisitor.h>
|
|
|
|
#include <Interpreters/Context.h>
|
|
|
|
#include <Interpreters/InDepthNodeVisitor.h>
|
|
|
|
#include <Interpreters/InterpreterDropQuery.h>
|
|
|
|
#include <Interpreters/InterpreterSelectQuery.h>
|
|
|
|
#include <Interpreters/getTableExpressions.h>
|
|
|
|
#include <Parsers/ASTCreateQuery.h>
|
|
|
|
#include <Parsers/ASTDropQuery.h>
|
|
|
|
#include <Parsers/ASTIdentifier.h>
|
|
|
|
#include <Parsers/ASTLiteral.h>
|
|
|
|
#include <Parsers/ASTSelectQuery.h>
|
|
|
|
#include <Parsers/ASTSubquery.h>
|
|
|
|
#include <Parsers/ASTTablesInSelectQuery.h>
|
|
|
|
#include <Parsers/ASTWatchQuery.h>
|
|
|
|
#include <Parsers/formatAST.h>
|
|
|
|
#include <Storages/StorageFactory.h>
|
|
|
|
#include <Common/typeid_cast.h>
|
|
|
|
|
|
|
|
#include <Storages/WindowView/StorageWindowView.h>
|
|
|
|
#include <Storages/WindowView/WindowViewBlockInputStream.h>
|
|
|
|
#include <Storages/WindowView/WindowViewBlocksBlockInputStream.h>
|
|
|
|
#include <Storages/WindowView/WindowViewProxyStorage.h>
|
|
|
|
|
|
|
|
|
|
|
|
namespace DB
|
|
|
|
{
|
|
|
|
namespace ErrorCodes
|
|
|
|
{
|
|
|
|
extern const int LOGICAL_ERROR;
|
|
|
|
extern const int INCORRECT_QUERY;
|
|
|
|
extern const int TABLE_WAS_NOT_DROPPED;
|
|
|
|
extern const int QUERY_IS_NOT_SUPPORTED_IN_WINDOW_VIEW;
|
|
|
|
extern const int SUPPORT_IS_DISABLED;
|
|
|
|
extern const int NUMBER_OF_ARGUMENTS_DOESNT_MATCH;
|
|
|
|
}
|
|
|
|
|
|
|
|
namespace
|
|
|
|
{
|
|
|
|
const auto RESCHEDULE_MS = 500;
|
|
|
|
|
|
|
|
class ParserStageMergeableOneMatcher
|
|
|
|
{
|
|
|
|
public:
|
|
|
|
using Visitor = InDepthNodeVisitor<ParserStageMergeableOneMatcher, true>;
|
|
|
|
|
|
|
|
struct Data
|
|
|
|
{
|
|
|
|
ASTPtr window_function;
|
|
|
|
String window_column_name;
|
|
|
|
// String window_column_name_or_alias;
|
|
|
|
bool is_tumble = false;
|
|
|
|
bool is_hop = false;
|
|
|
|
};
|
|
|
|
|
|
|
|
static bool needChildVisit(ASTPtr & node, const ASTPtr &)
|
|
|
|
{
|
|
|
|
if (node->as<ASTFunction>())
|
|
|
|
return false;
|
|
|
|
return true;
|
|
|
|
}
|
|
|
|
|
|
|
|
static void visit(ASTPtr & ast, Data & data)
|
|
|
|
{
|
|
|
|
if (const auto * t = ast->as<ASTFunction>())
|
|
|
|
visit(*t, ast, data);
|
|
|
|
}
|
|
|
|
|
|
|
|
private:
|
|
|
|
static void visit(const ASTFunction & node, ASTPtr & node_ptr, Data & data)
|
|
|
|
{
|
|
|
|
if (node.name == "TUMBLE")
|
|
|
|
{
|
2020-01-14 09:53:52 +00:00
|
|
|
if (!data.window_function)
|
2020-01-14 03:07:31 +00:00
|
|
|
{
|
|
|
|
data.is_tumble = true;
|
|
|
|
data.window_column_name = node.getColumnName();
|
|
|
|
data.window_function = node.clone();
|
|
|
|
}
|
|
|
|
else if (serializeAST(node) != serializeAST(*data.window_function))
|
|
|
|
throw Exception("WINDOW VIEW only support ONE WINDOW FUNCTION", ErrorCodes::QUERY_IS_NOT_SUPPORTED_IN_WINDOW_VIEW);
|
|
|
|
}
|
|
|
|
else if (node.name == "HOP")
|
|
|
|
{
|
2020-01-14 09:53:52 +00:00
|
|
|
if (!data.window_function)
|
2020-01-14 03:07:31 +00:00
|
|
|
{
|
|
|
|
data.is_hop = true;
|
|
|
|
data.window_function = node.clone();
|
|
|
|
auto ptr_ = node.clone();
|
|
|
|
std::static_pointer_cast<ASTFunction>(ptr_)->setAlias("");
|
|
|
|
auto arrayJoin = makeASTFunction("arrayJoin", ptr_);
|
|
|
|
arrayJoin->alias = node.alias;
|
|
|
|
data.window_column_name = arrayJoin->getColumnName();
|
|
|
|
node_ptr = arrayJoin;
|
|
|
|
}
|
2020-01-14 09:53:52 +00:00
|
|
|
else if (serializeAST(node) != serializeAST(*data.window_function))
|
2020-01-14 03:07:31 +00:00
|
|
|
throw Exception("WINDOW VIEW only support ONE WINDOW FUNCTION", ErrorCodes::QUERY_IS_NOT_SUPPORTED_IN_WINDOW_VIEW);
|
|
|
|
}
|
|
|
|
}
|
|
|
|
};
|
|
|
|
}
|
|
|
|
|
|
|
|
static void extractDependentTable(ASTSelectQuery & query, String & select_database_name, String & select_table_name)
|
|
|
|
{
|
|
|
|
auto db_and_table = getDatabaseAndTable(query, 0);
|
|
|
|
ASTPtr subquery = extractTableExpression(query, 0);
|
|
|
|
|
|
|
|
if (!db_and_table && !subquery)
|
|
|
|
return;
|
|
|
|
|
|
|
|
if (db_and_table)
|
|
|
|
{
|
|
|
|
select_table_name = db_and_table->table;
|
|
|
|
|
|
|
|
if (db_and_table->database.empty())
|
|
|
|
{
|
|
|
|
db_and_table->database = select_database_name;
|
|
|
|
AddDefaultDatabaseVisitor visitor(select_database_name);
|
|
|
|
visitor.visit(query);
|
|
|
|
}
|
|
|
|
else
|
|
|
|
select_database_name = db_and_table->database;
|
|
|
|
}
|
|
|
|
else if (auto * ast_select = subquery->as<ASTSelectWithUnionQuery>())
|
|
|
|
{
|
|
|
|
if (ast_select->list_of_selects->children.size() != 1)
|
|
|
|
throw Exception("UNION is not supported for WINDOW VIEW", ErrorCodes::QUERY_IS_NOT_SUPPORTED_IN_WINDOW_VIEW);
|
|
|
|
|
|
|
|
auto & inner_query = ast_select->list_of_selects->children.at(0);
|
|
|
|
|
|
|
|
extractDependentTable(inner_query->as<ASTSelectQuery &>(), select_database_name, select_table_name);
|
|
|
|
}
|
|
|
|
else
|
|
|
|
throw Exception(
|
|
|
|
"Logical error while creating StorageWindowView."
|
|
|
|
" Could not retrieve table name from select query.",
|
|
|
|
DB::ErrorCodes::LOGICAL_ERROR);
|
|
|
|
}
|
|
|
|
|
|
|
|
void StorageWindowView::checkTableCanBeDropped() const
|
|
|
|
{
|
2020-01-24 02:45:45 +00:00
|
|
|
auto table_id = getStorageID();
|
|
|
|
Dependencies dependencies = global_context.getDependencies(table_id);
|
2020-01-14 03:07:31 +00:00
|
|
|
if (!dependencies.empty())
|
|
|
|
{
|
2020-01-24 02:45:45 +00:00
|
|
|
StorageID dependent_table_id = dependencies.front();
|
|
|
|
throw Exception("Table has dependency " + dependent_table_id.getNameForLogs(), ErrorCodes::TABLE_WAS_NOT_DROPPED);
|
2020-01-14 03:07:31 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
void StorageWindowView::drop(TableStructureWriteLockHolder &)
|
|
|
|
{
|
2020-01-24 02:45:45 +00:00
|
|
|
auto table_id = getStorageID();
|
|
|
|
global_context.removeDependency(select_table_id, table_id);
|
2020-01-14 03:07:31 +00:00
|
|
|
|
|
|
|
std::lock_guard lock(mutex);
|
|
|
|
is_dropped = true;
|
|
|
|
condition.notify_all();
|
|
|
|
}
|
|
|
|
|
|
|
|
inline void StorageWindowView::flushToTable()
|
|
|
|
{
|
|
|
|
//write into dependent table
|
|
|
|
StoragePtr target_table = getTargetTable();
|
|
|
|
auto _blockInputStreamPtr = getNewBlocksInputStreamPtr();
|
|
|
|
auto _lock = target_table->lockStructureForShare(true, global_context.getCurrentQueryId());
|
|
|
|
auto stream = target_table->write(getInnerQuery(), global_context);
|
|
|
|
copyData(*_blockInputStreamPtr, *stream);
|
|
|
|
}
|
|
|
|
|
|
|
|
UInt32 StorageWindowView::getWindowUpperBound(UInt32 time_sec)
|
|
|
|
{
|
|
|
|
switch (window_kind)
|
|
|
|
{
|
|
|
|
#define CASE_WINDOW_KIND(KIND) \
|
|
|
|
case IntervalKind::KIND: \
|
|
|
|
{ \
|
|
|
|
UInt32 start = ToStartOfTransform<IntervalKind::KIND>::execute(time_sec, window_num_units, time_zone); \
|
|
|
|
return AddTime<IntervalKind::KIND>::execute(start, window_num_units, time_zone); \
|
|
|
|
}
|
2020-01-14 16:24:26 +00:00
|
|
|
CASE_WINDOW_KIND(Second)
|
|
|
|
CASE_WINDOW_KIND(Minute)
|
|
|
|
CASE_WINDOW_KIND(Hour)
|
|
|
|
CASE_WINDOW_KIND(Day)
|
|
|
|
CASE_WINDOW_KIND(Week)
|
|
|
|
CASE_WINDOW_KIND(Month)
|
|
|
|
CASE_WINDOW_KIND(Quarter)
|
|
|
|
CASE_WINDOW_KIND(Year)
|
2020-01-14 03:07:31 +00:00
|
|
|
#undef CASE_WINDOW_KIND
|
|
|
|
}
|
|
|
|
__builtin_unreachable();
|
|
|
|
}
|
|
|
|
|
|
|
|
void StorageWindowView::threadFuncToTable()
|
|
|
|
{
|
2020-01-24 02:45:45 +00:00
|
|
|
while (!shutdown_called && !target_table_id.empty())
|
2020-01-14 03:07:31 +00:00
|
|
|
{
|
|
|
|
std::unique_lock lock(flushTableMutex);
|
|
|
|
UInt64 timestamp_usec = static_cast<UInt64>(Poco::Timestamp().epochMicroseconds());
|
2020-01-14 09:04:47 +00:00
|
|
|
UInt64 w_end = static_cast<UInt64>(getWindowUpperBound(static_cast<UInt32>(timestamp_usec / 1000000))) * 1000000;
|
2020-01-14 03:07:31 +00:00
|
|
|
condition.wait_for(lock, std::chrono::microseconds(w_end - timestamp_usec));
|
|
|
|
try
|
|
|
|
{
|
|
|
|
if (refreshBlockStatus())
|
|
|
|
flushToTable();
|
|
|
|
}
|
|
|
|
catch (...)
|
|
|
|
{
|
|
|
|
tryLogCurrentException(__PRETTY_FUNCTION__);
|
|
|
|
break;
|
|
|
|
}
|
|
|
|
}
|
|
|
|
if (!shutdown_called)
|
|
|
|
toTableTask->scheduleAfter(RESCHEDULE_MS);
|
|
|
|
}
|
|
|
|
|
|
|
|
BlockInputStreams StorageWindowView::watch(
|
|
|
|
const Names & /*column_names*/,
|
|
|
|
const SelectQueryInfo & query_info,
|
|
|
|
const Context & context,
|
|
|
|
QueryProcessingStage::Enum & processed_stage,
|
|
|
|
size_t /*max_block_size*/,
|
|
|
|
const unsigned /*num_streams*/)
|
|
|
|
{
|
2020-01-24 02:45:45 +00:00
|
|
|
if (!target_table_id.empty())
|
2020-01-14 03:07:31 +00:00
|
|
|
throw Exception("WATCH query is disabled for " + getName() + " when constructed with 'TO' clause.", ErrorCodes::INCORRECT_QUERY);
|
|
|
|
|
2020-01-14 09:53:52 +00:00
|
|
|
if (active_ptr.use_count() > 1)
|
2020-01-14 03:07:31 +00:00
|
|
|
throw Exception("WATCH query is already attached, WINDOW VIEW only supports attaching one watch query.", ErrorCodes::INCORRECT_QUERY);
|
|
|
|
|
|
|
|
ASTWatchQuery & query = typeid_cast<ASTWatchQuery &>(*query_info.query);
|
|
|
|
|
|
|
|
bool has_limit = false;
|
|
|
|
UInt64 limit = 0;
|
|
|
|
|
|
|
|
if (query.limit_length)
|
|
|
|
{
|
|
|
|
has_limit = true;
|
|
|
|
limit = safeGet<UInt64>(typeid_cast<ASTLiteral &>(*query.limit_length).value);
|
|
|
|
}
|
|
|
|
|
|
|
|
auto reader = std::make_shared<WindowViewBlockInputStream>(
|
|
|
|
std::static_pointer_cast<StorageWindowView>(shared_from_this()),
|
|
|
|
active_ptr,
|
|
|
|
has_limit,
|
|
|
|
limit,
|
|
|
|
context.getSettingsRef().temporary_window_view_timeout.totalSeconds());
|
|
|
|
|
|
|
|
{
|
|
|
|
std::lock_guard no_users_thread_lock(no_users_thread_mutex);
|
|
|
|
if (no_users_thread.joinable())
|
|
|
|
{
|
|
|
|
std::lock_guard lock(no_users_thread_wakeup_mutex);
|
|
|
|
no_users_thread_wakeup = true;
|
|
|
|
no_users_thread_condition.notify_one();
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
processed_stage = QueryProcessingStage::Complete;
|
|
|
|
|
|
|
|
return {reader};
|
|
|
|
}
|
|
|
|
|
|
|
|
Block StorageWindowView::getHeader() const
|
|
|
|
{
|
|
|
|
if (!sample_block)
|
|
|
|
{
|
2020-01-24 02:45:45 +00:00
|
|
|
auto storage = global_context.getTable(select_table_id);
|
|
|
|
sample_block = InterpreterSelectQuery(getInnerQuery(), global_context, storage, SelectQueryOptions(QueryProcessingStage::Complete))
|
2020-01-14 03:07:31 +00:00
|
|
|
.getSampleBlock();
|
|
|
|
for (size_t i = 0; i < sample_block.columns(); ++i)
|
|
|
|
sample_block.safeGetByPosition(i).column = sample_block.safeGetByPosition(i).column->convertToFullColumnIfConst();
|
|
|
|
}
|
|
|
|
|
|
|
|
return sample_block;
|
|
|
|
}
|
|
|
|
|
|
|
|
StorageWindowView::StorageWindowView(
|
2020-01-24 02:45:45 +00:00
|
|
|
const StorageID & table_id_,
|
2020-01-14 03:07:31 +00:00
|
|
|
Context & local_context,
|
|
|
|
const ASTCreateQuery & query,
|
|
|
|
const ColumnsDescription & columns_)
|
2020-01-24 02:45:45 +00:00
|
|
|
: IStorage(table_id_)
|
2020-01-14 03:07:31 +00:00
|
|
|
, global_context(local_context.getGlobalContext())
|
|
|
|
, time_zone(DateLUT::instance())
|
|
|
|
{
|
|
|
|
setColumns(columns_);
|
|
|
|
|
|
|
|
if (!query.select)
|
|
|
|
throw Exception("SELECT query is not specified for " + getName(), ErrorCodes::INCORRECT_QUERY);
|
|
|
|
|
|
|
|
/// Default value, if only table name exist in the query
|
|
|
|
if (query.select->list_of_selects->children.size() != 1)
|
|
|
|
throw Exception("UNION is not supported for Window View", ErrorCodes::QUERY_IS_NOT_SUPPORTED_IN_WINDOW_VIEW);
|
|
|
|
|
|
|
|
auto inner_query_ = query.select->list_of_selects->children.at(0);
|
|
|
|
|
|
|
|
ASTSelectQuery & select_query = typeid_cast<ASTSelectQuery &>(*inner_query_);
|
2020-01-24 02:45:45 +00:00
|
|
|
String select_database_name = local_context.getCurrentDatabase();
|
|
|
|
String select_table_name;
|
2020-01-14 03:07:31 +00:00
|
|
|
extractDependentTable(select_query, select_database_name, select_table_name);
|
2020-01-24 02:45:45 +00:00
|
|
|
select_table_id = StorageID(select_database_name, select_table_name);
|
2020-01-14 03:07:31 +00:00
|
|
|
inner_query = innerQueryParser(select_query);
|
|
|
|
|
|
|
|
/// If the table is not specified - use the table `system.one`
|
|
|
|
if (select_table_name.empty())
|
|
|
|
{
|
|
|
|
select_database_name = "system";
|
|
|
|
select_table_name = "one";
|
|
|
|
}
|
|
|
|
|
2020-01-24 02:45:45 +00:00
|
|
|
global_context.addDependency(select_table_id, table_id_);
|
|
|
|
|
|
|
|
parent_storage = local_context.getTable(select_table_id);
|
2020-01-14 03:07:31 +00:00
|
|
|
|
|
|
|
if (!query.to_table.empty())
|
|
|
|
{
|
2020-01-24 02:45:45 +00:00
|
|
|
target_table_id = StorageID(query.to_database, query.to_table);
|
2020-01-14 03:07:31 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
is_temporary = query.temporary;
|
|
|
|
temporary_window_view_timeout = local_context.getSettingsRef().temporary_window_view_timeout.totalSeconds();
|
|
|
|
|
|
|
|
mergeable_blocks = std::make_shared<std::list<BlocksListPtr>>();
|
|
|
|
|
|
|
|
active_ptr = std::make_shared<bool>(true);
|
|
|
|
|
2020-01-24 02:45:45 +00:00
|
|
|
toTableTask = global_context.getSchedulePool().createTask(getStorageID().getFullTableName(), [this] { threadFuncToTable(); });
|
2020-01-14 03:07:31 +00:00
|
|
|
toTableTask->deactivate();
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
ASTPtr StorageWindowView::innerQueryParser(ASTSelectQuery & query)
|
|
|
|
{
|
|
|
|
if (!query.groupBy())
|
|
|
|
throw Exception("GROUP BY query is required for " + getName(), ErrorCodes::INCORRECT_QUERY);
|
|
|
|
|
|
|
|
// parse stage mergeable
|
|
|
|
ASTPtr result = query.clone();
|
|
|
|
ASTPtr expr_list = result;
|
|
|
|
ParserStageMergeableOneMatcher::Data stageMergeableOneData;
|
|
|
|
ParserStageMergeableOneMatcher::Visitor(stageMergeableOneData).visit(expr_list);
|
|
|
|
if (!stageMergeableOneData.is_tumble && !stageMergeableOneData.is_hop)
|
|
|
|
throw Exception("WINDOW FUNCTION is not specified for " + getName(), ErrorCodes::INCORRECT_QUERY);
|
|
|
|
window_column_name = stageMergeableOneData.window_column_name;
|
|
|
|
|
|
|
|
// parser window function
|
|
|
|
ASTFunction & window_function = typeid_cast<ASTFunction &>(*stageMergeableOneData.window_function);
|
|
|
|
ASTExpressionList &window_function_args = typeid_cast<ASTExpressionList&>(*window_function.arguments);
|
|
|
|
const auto & children = window_function_args.children;
|
|
|
|
const auto & interval_p1 = std::static_pointer_cast<ASTFunction>(children.at(1));
|
2020-01-14 09:53:52 +00:00
|
|
|
if (!interval_p1 || !startsWith(interval_p1->name, "toInterval"))
|
2020-01-14 03:07:31 +00:00
|
|
|
throw Exception("Illegal type of last argument of function " + interval_p1->name + " should be Interval", ErrorCodes::ILLEGAL_COLUMN);
|
|
|
|
|
|
|
|
String interval_str = interval_p1->name.substr(10);
|
2020-01-14 09:53:52 +00:00
|
|
|
if (interval_str == "Second")
|
2020-01-14 03:07:31 +00:00
|
|
|
window_kind = IntervalKind::Second;
|
2020-01-14 09:53:52 +00:00
|
|
|
else if (interval_str == "Minute")
|
2020-01-14 03:07:31 +00:00
|
|
|
window_kind = IntervalKind::Minute;
|
2020-01-14 09:53:52 +00:00
|
|
|
else if (interval_str == "Hour")
|
2020-01-14 03:07:31 +00:00
|
|
|
window_kind = IntervalKind::Hour;
|
2020-01-14 09:53:52 +00:00
|
|
|
else if (interval_str == "Day")
|
2020-01-14 03:07:31 +00:00
|
|
|
window_kind = IntervalKind::Day;
|
2020-01-14 09:53:52 +00:00
|
|
|
else if (interval_str == "Week")
|
2020-01-14 03:07:31 +00:00
|
|
|
window_kind = IntervalKind::Week;
|
2020-01-14 09:53:52 +00:00
|
|
|
else if (interval_str == "Month")
|
2020-01-14 03:07:31 +00:00
|
|
|
window_kind = IntervalKind::Month;
|
2020-01-14 09:53:52 +00:00
|
|
|
else if (interval_str == "Quarter")
|
2020-01-14 03:07:31 +00:00
|
|
|
window_kind = IntervalKind::Quarter;
|
2020-01-14 09:53:52 +00:00
|
|
|
else if (interval_str == "Year")
|
2020-01-14 03:07:31 +00:00
|
|
|
window_kind = IntervalKind::Year;
|
|
|
|
|
|
|
|
const auto & interval_units_p1 = std::static_pointer_cast<ASTLiteral>(interval_p1->children.front()->children.front());
|
|
|
|
|
|
|
|
window_num_units = stoi(interval_units_p1->value.get<String>());
|
|
|
|
return result;
|
|
|
|
}
|
|
|
|
|
|
|
|
void StorageWindowView::writeIntoWindowView(StorageWindowView & window_view, const Block & block, const Context & context)
|
|
|
|
{
|
|
|
|
BlockInputStreams streams = {std::make_shared<OneBlockInputStream>(block)};
|
2020-01-24 02:45:45 +00:00
|
|
|
auto window_proxy_storage = std::make_shared<WindowViewProxyStorage>(StorageID("", "WindowViewProxyStorage"),
|
2020-01-14 03:07:31 +00:00
|
|
|
window_view.getParentStorage(), std::move(streams), QueryProcessingStage::FetchColumns);
|
|
|
|
InterpreterSelectQuery select_block(
|
|
|
|
window_view.getInnerQuery(), context, window_proxy_storage, QueryProcessingStage::WithMergeableState);
|
|
|
|
|
|
|
|
auto data_mergeable_stream = std::make_shared<MaterializingBlockInputStream>(select_block.execute().in);
|
|
|
|
|
|
|
|
BlocksListPtr new_mergeable_blocks = std::make_shared<BlocksList>();
|
|
|
|
while (Block block_ = data_mergeable_stream->read())
|
|
|
|
{
|
|
|
|
const ColumnTuple * column_tuple = checkAndGetColumn<ColumnTuple>(block_.getByName(window_view.window_column_name).column.get());
|
|
|
|
block_.insert(
|
|
|
|
{ColumnUInt8::create(block_.rows(), WINDOW_VIEW_FIRE_STATUS::WAITING), std::make_shared<DataTypeUInt8>(), "____fire_status"});
|
|
|
|
block_.insert({column_tuple->getColumnPtr(1), std::make_shared<DataTypeDateTime>(), "____w_end"});
|
|
|
|
new_mergeable_blocks->push_back(block_);
|
|
|
|
}
|
|
|
|
|
|
|
|
if (new_mergeable_blocks->empty())
|
|
|
|
return;
|
|
|
|
|
|
|
|
{
|
|
|
|
std::unique_lock lock(window_view.mutex);
|
|
|
|
window_view.getMergeableBlocksList()->push_back(new_mergeable_blocks);
|
|
|
|
}
|
|
|
|
window_view.condition.notify_all();
|
|
|
|
}
|
|
|
|
|
|
|
|
StoragePtr StorageWindowView::getTargetTable() const
|
|
|
|
{
|
2020-01-24 02:45:45 +00:00
|
|
|
return global_context.getTable(target_table_id);
|
2020-01-14 03:07:31 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
StoragePtr StorageWindowView::tryGetTargetTable() const
|
|
|
|
{
|
2020-01-24 02:45:45 +00:00
|
|
|
return global_context.tryGetTable(target_table_id);
|
2020-01-14 03:07:31 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
void StorageWindowView::startup()
|
|
|
|
{
|
|
|
|
// Start the working thread
|
2020-01-24 02:45:45 +00:00
|
|
|
if (!target_table_id.empty())
|
2020-01-14 03:07:31 +00:00
|
|
|
toTableTask->activateAndSchedule();
|
|
|
|
startNoUsersThread(temporary_window_view_timeout);
|
|
|
|
}
|
|
|
|
|
|
|
|
void StorageWindowView::shutdown()
|
|
|
|
{
|
|
|
|
bool expected = false;
|
|
|
|
if (!shutdown_called.compare_exchange_strong(expected, true))
|
|
|
|
return;
|
|
|
|
toTableTask->deactivate();
|
|
|
|
{
|
|
|
|
std::lock_guard no_users_thread_lock(no_users_thread_mutex);
|
|
|
|
if (no_users_thread.joinable())
|
|
|
|
{
|
|
|
|
{
|
|
|
|
std::lock_guard lock(no_users_thread_wakeup_mutex);
|
|
|
|
no_users_thread_wakeup = true;
|
|
|
|
no_users_thread_condition.notify_one();
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
StorageWindowView::~StorageWindowView()
|
|
|
|
{
|
|
|
|
shutdown();
|
|
|
|
|
|
|
|
{
|
|
|
|
std::lock_guard lock(no_users_thread_mutex);
|
|
|
|
if (no_users_thread.joinable())
|
|
|
|
no_users_thread.detach();
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
bool StorageWindowView::refreshBlockStatus()
|
|
|
|
{
|
|
|
|
UInt32 timestamp_now = std::time(nullptr);
|
|
|
|
for (BlocksListPtr mergeable_block : *mergeable_blocks)
|
|
|
|
{
|
|
|
|
for (Block & block : *mergeable_block)
|
|
|
|
{
|
|
|
|
auto & col_wend = block.getByName("____w_end").column;
|
|
|
|
const auto & wend_data = static_cast<const ColumnUInt32 &>(*col_wend).getData();
|
|
|
|
auto & col_status = block.getByName("____fire_status").column;
|
|
|
|
const auto & col_status_data = static_cast<const ColumnUInt8 &>(*col_status).getData();
|
|
|
|
|
|
|
|
for (size_t i = 0; i < col_wend->size(); ++i)
|
|
|
|
{
|
|
|
|
if (wend_data[i] < timestamp_now && col_status_data[i] == WINDOW_VIEW_FIRE_STATUS::WAITING)
|
|
|
|
return true;
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
return false;
|
|
|
|
}
|
|
|
|
|
|
|
|
BlockInputStreamPtr StorageWindowView::getNewBlocksInputStreamPtr()
|
|
|
|
{
|
|
|
|
if (mergeable_blocks->empty())
|
|
|
|
return {std::make_shared<NullBlockInputStream>(getHeader())};
|
|
|
|
{
|
|
|
|
std::lock_guard lock(mutex);
|
|
|
|
//delete fired blocks
|
|
|
|
for (BlocksListPtr mergeable_block : *mergeable_blocks)
|
|
|
|
{
|
2020-01-14 09:53:52 +00:00
|
|
|
mergeable_block->remove_if([](Block & block_)
|
|
|
|
{
|
2020-01-14 03:07:31 +00:00
|
|
|
auto & column_ = block_.getByName("____fire_status").column;
|
|
|
|
const auto & data = static_cast<const ColumnUInt8 &>(*column_).getData();
|
|
|
|
for (size_t i = 0; i < column_->size(); ++i)
|
|
|
|
{
|
|
|
|
if (data[i] != WINDOW_VIEW_FIRE_STATUS::RETIRED)
|
|
|
|
return false;
|
|
|
|
}
|
|
|
|
return true;
|
|
|
|
});
|
|
|
|
}
|
|
|
|
mergeable_blocks->remove_if([](BlocksListPtr & ptr) { return ptr->size() == 0; });
|
|
|
|
|
|
|
|
if (mergeable_blocks->empty())
|
|
|
|
return {std::make_shared<NullBlockInputStream>(getHeader())};
|
|
|
|
|
|
|
|
// mark blocks can be fired
|
|
|
|
UInt32 timestamp_now = std::time(nullptr);
|
|
|
|
for (BlocksListPtr mergeable_block : *mergeable_blocks)
|
|
|
|
{
|
|
|
|
for (Block & block : *mergeable_block)
|
|
|
|
{
|
|
|
|
auto & col_wend = block.getByName("____w_end").column;
|
|
|
|
const auto & wend_data = static_cast<const ColumnUInt32 &>(*col_wend).getData();
|
|
|
|
auto & col_status = block.getByName("____fire_status").column;
|
|
|
|
auto col_status_mutable = col_status->assumeMutable();
|
|
|
|
auto & col_status_data = static_cast<ColumnUInt8 &>(*col_status_mutable).getData();
|
|
|
|
|
|
|
|
for (size_t i = 0; i < col_wend->size(); ++i)
|
|
|
|
{
|
|
|
|
if (wend_data[i] < timestamp_now && col_status_data[i] == WINDOW_VIEW_FIRE_STATUS::WAITING)
|
|
|
|
col_status_data[i] = WINDOW_VIEW_FIRE_STATUS::READY;
|
|
|
|
}
|
|
|
|
col_status = std::move(col_status_mutable);
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
BlockInputStreams from;
|
|
|
|
auto sample_block_ = mergeable_blocks->front()->front().cloneEmpty(); //TODO: 改为全局
|
|
|
|
BlockInputStreamPtr stream = std::make_shared<WindowViewBlocksBlockInputStream>(mergeable_blocks, sample_block_, mutex);
|
|
|
|
from.push_back(std::move(stream));
|
|
|
|
auto proxy_storage = std::make_shared<WindowViewProxyStorage>(
|
2020-01-24 02:45:45 +00:00
|
|
|
StorageID("", "WindowViewProxyStorage"), getParentStorage(), std::move(from), QueryProcessingStage::WithMergeableState);
|
2020-01-14 03:07:31 +00:00
|
|
|
InterpreterSelectQuery select(getInnerQuery(), global_context, proxy_storage, QueryProcessingStage::Complete);
|
|
|
|
BlockInputStreamPtr data = std::make_shared<MaterializingBlockInputStream>(select.execute().in);
|
|
|
|
return data;
|
|
|
|
}
|
|
|
|
|
|
|
|
BlocksPtr StorageWindowView::getNewBlocks()
|
|
|
|
{
|
|
|
|
auto res = getNewBlocksInputStreamPtr();
|
|
|
|
BlocksPtr blocks = std::make_shared<Blocks>();
|
|
|
|
while (Block this_block = res->read())
|
|
|
|
blocks->push_back(this_block);
|
|
|
|
return blocks;
|
|
|
|
}
|
|
|
|
|
|
|
|
void StorageWindowView::noUsersThread(std::shared_ptr<StorageWindowView> storage, const UInt64 & timeout)
|
|
|
|
{
|
|
|
|
bool drop_table = false;
|
|
|
|
|
|
|
|
if (storage->shutdown_called)
|
|
|
|
return;
|
|
|
|
|
2020-01-24 02:45:45 +00:00
|
|
|
auto table_id = storage->getStorageID();
|
2020-01-14 03:07:31 +00:00
|
|
|
{
|
|
|
|
while (1)
|
|
|
|
{
|
|
|
|
std::unique_lock lock(storage->no_users_thread_wakeup_mutex);
|
|
|
|
if (!storage->no_users_thread_condition.wait_for(lock, std::chrono::seconds(timeout), [&] { return storage->no_users_thread_wakeup; }))
|
|
|
|
{
|
|
|
|
storage->no_users_thread_wakeup = false;
|
|
|
|
if (storage->shutdown_called)
|
|
|
|
return;
|
2020-01-24 02:45:45 +00:00
|
|
|
if (!storage->global_context.getDependencies(table_id).empty())
|
2020-01-14 03:07:31 +00:00
|
|
|
continue;
|
|
|
|
drop_table = true;
|
|
|
|
}
|
|
|
|
break;
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
if (drop_table)
|
|
|
|
{
|
2020-01-24 02:45:45 +00:00
|
|
|
if (storage->global_context.tryGetTable(table_id))
|
2020-01-14 03:07:31 +00:00
|
|
|
{
|
|
|
|
try
|
|
|
|
{
|
|
|
|
/// We create and execute `drop` query for this table
|
|
|
|
auto drop_query = std::make_shared<ASTDropQuery>();
|
2020-01-24 02:45:45 +00:00
|
|
|
drop_query->database = table_id.database_name;
|
|
|
|
drop_query->table = table_id.table_name;
|
2020-01-14 03:07:31 +00:00
|
|
|
drop_query->kind = ASTDropQuery::Kind::Drop;
|
|
|
|
ASTPtr ast_drop_query = drop_query;
|
|
|
|
InterpreterDropQuery drop_interpreter(ast_drop_query, storage->global_context);
|
|
|
|
drop_interpreter.execute();
|
|
|
|
}
|
|
|
|
catch (...)
|
|
|
|
{
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
void StorageWindowView::startNoUsersThread(const UInt64 & timeout)
|
|
|
|
{
|
|
|
|
bool expected = false;
|
|
|
|
if (!start_no_users_thread_called.compare_exchange_strong(expected, true))
|
|
|
|
return;
|
|
|
|
|
|
|
|
if (is_temporary)
|
|
|
|
{
|
|
|
|
std::lock_guard no_users_thread_lock(no_users_thread_mutex);
|
|
|
|
|
|
|
|
if (shutdown_called)
|
|
|
|
return;
|
|
|
|
|
|
|
|
if (no_users_thread.joinable())
|
|
|
|
{
|
|
|
|
{
|
|
|
|
std::lock_guard lock(no_users_thread_wakeup_mutex);
|
|
|
|
no_users_thread_wakeup = true;
|
|
|
|
no_users_thread_condition.notify_one();
|
|
|
|
}
|
|
|
|
no_users_thread.join();
|
|
|
|
}
|
|
|
|
{
|
|
|
|
std::lock_guard lock(no_users_thread_wakeup_mutex);
|
|
|
|
no_users_thread_wakeup = false;
|
|
|
|
}
|
|
|
|
if (!is_dropped)
|
|
|
|
no_users_thread = std::thread(&StorageWindowView::noUsersThread,
|
|
|
|
std::static_pointer_cast<StorageWindowView>(shared_from_this()), timeout);
|
|
|
|
}
|
|
|
|
|
|
|
|
start_no_users_thread_called = false;
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
void registerStorageWindowView(StorageFactory & factory)
|
|
|
|
{
|
2020-01-14 09:53:52 +00:00
|
|
|
factory.registerStorage("WindowView", [](const StorageFactory::Arguments & args)
|
|
|
|
{
|
2020-01-14 03:07:31 +00:00
|
|
|
if (!args.attach && !args.local_context.getSettingsRef().allow_experimental_window_view)
|
|
|
|
throw Exception(
|
|
|
|
"Experimental WINDOW VIEW feature is not enabled (the setting 'allow_experimental_window_view')",
|
|
|
|
ErrorCodes::SUPPORT_IS_DISABLED);
|
2020-01-24 02:45:45 +00:00
|
|
|
|
|
|
|
return StorageWindowView::create(args.table_id, args.local_context, args.query, args.columns);
|
2020-01-14 03:07:31 +00:00
|
|
|
});
|
|
|
|
}
|
2020-01-14 16:24:26 +00:00
|
|
|
}
|