Method sendExternalTables

src/QueryPipeline/RemoteQueryExecutor.cpp:877–948

}

void RemoteQueryExecutor::sendExternalTables()
{
    size_t count = connections->size();

    {
        std::lock_guard lock(external_tables_mutex);

        external_tables_data.clear();
        external_tables_data.reserve(count);

        StreamLocalLimits limits;
        const auto & settings = context->getSettingsRef();
        limits.mode = LimitsMode::LIMITS_TOTAL;
        limits.speed_limits.max_execution_time = settings[Setting::max_execution_time];
        limits.timeout_overflow_mode = settings[Setting::timeout_overflow_mode];
        limits.speed_limits.max_estimated_execution_time = settings[Setting::max_estimated_execution_time];

        for (size_t i = 0; i < count; ++i)
        {
            ExternalTablesData res;
            for (const auto & table : external_tables)
            {
                StoragePtr cur = table.second;
                /// Send only temporary tables with StorageMemory
                auto storage_memory = std::dynamic_pointer_cast<StorageMemory>(cur);
                if (!storage_memory)
                    continue;

                /// Skip sending Materialized CTEs when they are not built.
                /// It is required to be able CTE materialization plan with parallel replicas (avoiding
                /// circular dependency between CTE materialization and parallel replicas external tables.
                auto materialized_cte = storage_memory->getMaterializedCTE();
                if (materialized_cte != nullptr && !materialized_cte->isBuilt())
                {
                    LOG_DEBUG(log, "Skipping sending CTE '{}' because it has not been materialized yet", materialized_cte->cte_name);
                    continue;
                }

                auto data = std::make_unique<ExternalTableData>();
                data->table_name = table.first;
                data->creating_pipe_callback = [cur, limits, my_context = this->context]()
                {
                    SelectQueryInfo query_info;
                    auto metadata_snapshot = cur->getInMemoryMetadataPtr(my_context, false);
                    auto storage_snapshot = cur->getStorageSnapshot(metadata_snapshot, my_context);
                    QueryProcessingStage::Enum read_from_table_stage = cur->getQueryProcessingStage(
                        my_context, QueryProcessingStage::Complete, storage_snapshot, query_info);

                    QueryPlan plan;
                    cur->read(
                        plan,
                        metadata_snapshot->getColumns().getNamesOfPhysical(),
                        storage_snapshot, query_info, my_context,
                        read_from_table_stage, DEFAULT_BLOCK_SIZE, 1);

                    auto builder = plan.buildQueryPipeline(QueryPlanOptimizationSettings(my_context), BuildQueryPipelineSettings(my_context));

                    builder->resize(1);