Method StorageWindowView

src/Storages/WindowView/StorageWindowView.cpp:1252–1345

}

StorageWindowView::StorageWindowView(
    const StorageID & table_id_,
    ContextPtr context_,
    const ASTCreateQuery & query,
    const ColumnsDescription & columns_,
    const String & comment,
    LoadingStrictnessLevel mode)
    : IStorage(table_id_)
    , WithContext(context_->getGlobalContext())
    , log(getLogger(fmt::format("StorageWindowView({}.{})", table_id_.database_name, table_id_.table_name)))
    , fire_signal_timeout_s(context_->getSettingsRef()[Setting::wait_for_window_view_fire_signal_timeout].totalSeconds())
    , clean_interval_usec(context_->getSettingsRef()[Setting::window_view_clean_interval].totalMicroseconds())
{
    if (context_->getSettingsRef()[Setting::allow_experimental_analyzer])
        disabled_due_to_analyzer = true;

    if (mode <= LoadingStrictnessLevel::CREATE)
        throwIfWindowViewIsDisabled();

    if (!query.select)
        throw Exception(ErrorCodes::INCORRECT_QUERY, "SELECT query is not specified for {}", getName());

    StorageInMemoryMetadata storage_metadata;
    storage_metadata.setColumns(columns_);
    storage_metadata.setComment(comment);
    setInMemoryMetadata(storage_metadata);

    /// If the target table is not set, use inner target table
    auto to_table_id = query.getTargetTableID(ViewTarget::To);
    has_inner_target_table = to_table_id.empty();
    auto * to_table_engine = query.getTargetInnerEngine(ViewTarget::To);

    if (has_inner_target_table && !to_table_engine)
        throw Exception(ErrorCodes::INCORRECT_QUERY,
                        "You must specify where to save results of a WindowView query: "
                        "either ENGINE or an existing table in a TO clause");

    if (query.select->list_of_selects->children.size() != 1)
        throw Exception(
            ErrorCodes::QUERY_IS_NOT_SUPPORTED_IN_WINDOW_VIEW,
            "UNION is not supported for {}", getName());

    /// Extract information about watermark, lateness.
    eventTimeParser(query);

    auto inner_query = initInnerQuery(query.select->list_of_selects->children.at(0)->as<ASTSelectQuery &>(), context_);

    if (auto * inner_storage = query.getTargetInnerEngine(ViewTarget::Inner))
        inner_table_engine = inner_storage->clone();
    inner_table_id = StorageID(getStorageID().database_name, generateInnerTableName(getStorageID()));
    inner_fetch_query = generateInnerFetchQuery(inner_table_id);

    target_table_id = has_inner_target_table ? StorageID(table_id_.database_name, generateTargetTableName(table_id_)) : to_table_id;

    if (is_proctime)
        next_fire_signal = getWindowUpperBound(now());

    std::exchange(has_inner_table, true);