Method process

src/DataStreams/PushingToViewsBlockOutputStream.cpp:482–568

}

void PushingToViewsBlockOutputStream::process(const Block & block, ViewInfo & view)
{
    Stopwatch watch;

    try
    {
        BlockInputStreamPtr in;

        /// We need keep InterpreterSelectQuery, until the processing will be finished, since:
        ///
        /// - We copy Context inside InterpreterSelectQuery to support
        ///   modification of context (Settings) for subqueries
        /// - InterpreterSelectQuery lives shorter than query pipeline.
        ///   It's used just to build the query pipeline and no longer needed
        /// - ExpressionAnalyzer and then, Functions, that created in InterpreterSelectQuery,
        ///   **can** take a reference to Context from InterpreterSelectQuery
        ///   (the problem raises only when function uses context from the
        ///    execute*() method, like FunctionDictGet do)
        /// - These objects live inside query pipeline (DataStreams) and the reference become dangling.

        if (view.query)
        {
            /// We create a table with the same name as original table and the same alias columns,
            ///  but it will contain single block (that is INSERT-ed into main table).
            /// InterpreterSelectQuery will do processing of alias columns.

            auto local_context = Context::createCopy(select_context);
            local_context->addViewSource(
                StorageValues::create(storage->getStorageID(), metadata_snapshot->getColumns(), block, storage->getVirtuals()));

#if USE_RDKAFKA
            /// Set the limits for Kafka specially as Kafka stream may need long time to write a block with many map keys;
            /// If write data timeout, throw exception to trigger re-consumption
            if (auto * kafka = dynamic_cast<StorageCloudKafka *>(storage.get()))
            {
                auto settings = local_context->getSettingsRef();
                settings.max_execution_time = kafka->getSettings().max_write_execution_second;
                settings.timeout_overflow_mode = OverflowMode::THROW;
                local_context->setSettings(settings);
                LOG_TRACE(log, "Set max execution limit for reading from {} to write view with {} s",
                          storage->getStorageID().getNameForLogs(), local_context->getSettings().max_execution_time.totalSeconds());
            }
#endif

            if (auto * select = view.query->as<ASTSelectWithUnionQuery>())
            {
                InterpreterSelectWithUnionQuery interepter_select(view.query, local_context, SelectQueryOptions());
                in = std::make_shared<MaterializingBlockInputStream>(interepter_select.execute().getInputStream());
            }
            else
            {
                InterpreterSelectQuery interepter_select(view.query, local_context, SelectQueryOptions());
                in = std::make_shared<MaterializingBlockInputStream>(interepter_select.execute().getInputStream());
            }
            
            /// Squashing is needed here because the materialized view query can generate a lot of blocks
            /// even when only one block is inserted into the parent table (e.g. if the query is a GROUP BY
            /// and two-level aggregation is triggered).