Method buildQueryPipeline

src/Interpreters/InterpreterWatchQuery.cpp:67–116

}

QueryPipelineBuilder InterpreterWatchQuery::buildQueryPipeline()
{
    const ASTWatchQuery & query = typeid_cast<const ASTWatchQuery &>(*query_ptr);
    auto table_id = getContext()->resolveStorageID(query, Context::ResolveOrdinary);

    /// Get storage
    storage = DatabaseCatalog::instance().tryGetTable(table_id, getContext());

    if (!storage)
        throw Exception(ErrorCodes::UNKNOWN_TABLE, "Table {} does not exist.", table_id.getNameForLogs());

    auto storage_name = storage->getName();
    if (storage_name == "WindowView" && !getContext()->getSettingsRef()[Setting::allow_experimental_window_view])
        throw Exception(ErrorCodes::SUPPORT_IS_DISABLED,
                        "Experimental WINDOW VIEW feature is not enabled (the setting 'allow_experimental_window_view')");

    /// List of columns to read to execute the query.
    auto metadata_snapshot = storage->getInMemoryMetadataPtr(getContext(), false);
    Names required_columns = metadata_snapshot->getColumns().getNamesOfPhysical();
    getContext()->checkAccess(AccessType::SELECT, table_id, required_columns);

    /// Get context settings for this query
    const Settings & settings = getContext()->getSettingsRef();

    /// Limitation on the number of columns to read.
    if (settings[Setting::max_columns_to_read] && required_columns.size() > settings[Setting::max_columns_to_read])
        throw Exception(
            ErrorCodes::TOO_MANY_COLUMNS,
            "Limit for number of columns to read exceeded. "
            "Requested: {}, maximum: {}",
            required_columns.size(),
            settings[Setting::max_columns_to_read].toString());

    size_t max_block_size = settings[Setting::max_block_size];
    size_t max_streams = 1;

    /// Define query info
    SelectQueryInfo query_info;
    query_info.query = query_ptr;

    /// From stage
    QueryProcessingStage::Enum from_stage = QueryProcessingStage::FetchColumns;

    /// Watch storage
    auto pipe = storage->watch(required_columns, query_info, getContext(), from_stage, max_block_size, max_streams);

    QueryPipelineBuilder pipeline;
    pipeline.init(std::move(pipe));
    return pipeline;
}

void registerInterpreterWatchQuery(InterpreterFactory & factory);
void registerInterpreterWatchQuery(InterpreterFactory & factory)