{

if (!initialized())

throw Exception(ErrorCodes::LOGICAL_ERROR, "QueryPipeline is uninitialized");

{

checkInitialized();

if (pipe.isCompleted())

throw Exception(ErrorCodes::LOGICAL_ERROR, "QueryPipeline is already completed");

{

if (initialized())

throw Exception(ErrorCodes::LOGICAL_ERROR, "Pipeline has already been initialized");

if (pipe_.empty())

throw Exception(ErrorCodes::LOGICAL_ERROR, "Can't initialize pipeline with empty pipe");

pipe = std::move(pipe_);

{

Pipe pipe_to_destroy(std::move(pipe));

*this = QueryPipelineBuilder();

{

checkInitializedAndNotCompleted();

pipe.addSimpleTransform(getter);

{

checkInitializedAndNotCompleted();

pipe.addSimpleTransform(getter);

{

checkInitializedAndNotCompleted();

pipe.addTransform(std::move(transform));

{

checkInitializedAndNotCompleted();

pipe.addTransform(std::move(transform), totals, extremes);

{

checkInitializedAndNotCompleted();

pipe.addChains(std::move(chains));

{

checkInitializedAndNotCompleted();

VectorWithMemoryTracking<Chain> chains;

chains.emplace_back(std::move(chain));

pipe.resize(1);

pipe.addChains(std::move(chains));

{

checkInitializedAndNotCompleted();

pipe.transform(transformer, check_ports);

{

checkInitializedAndNotCompleted();

pipe.setSinks(getter);

AggregatingTransformParamsPtr params, size_t num_merging_processors, bool should_produce_results_in_order_of_bucket_number)

{

DB::addMergingAggregatedMemoryEfficientTransform(

pipe, std::move(params), num_merging_processors, should_produce_results_in_order_of_bucket_number);

{

checkInitializedAndNotCompleted();

pipe.resize(num_streams, strict, min_outstreams_per_resize_after_split);

{

checkInitializedAndNotCompleted();

narrowPipe(pipe, size);

{

checkInitializedAndNotCompleted();

if (!typeid_cast<const TotalsHavingTransform *>(transform.get()))

throw Exception(ErrorCodes::LOGICAL_ERROR, "TotalsHavingTransform is expected for QueryPipeline::addTotalsHavingTransform");

if (pipe.getTotalsPort())

throw Exception(ErrorCodes::LOGICAL_ERROR, "Totals having transform was already added to pipeline");

resize(1);

auto * totals_port = &transform->getOutputs().back();

pipe.addTransform(std::move(transform), totals_port, nullptr);

{

checkInitializedAndNotCompleted();

if (pipe.getTotalsPort())

throw Exception(ErrorCodes::LOGICAL_ERROR, "Totals having transform was already added to pipeline");

const auto & current_header = getSharedHeader();

Columns columns;

columns.reserve(current_header->columns());

for (size_t i = 0; i < current_header->columns(); ++i)

{

const auto & elem = current_header->getByPosition(i);

/// Keep the value of a constant column in the synthesized totals row.

/// Every row of a constant column carries the same value, so the totals

/// row must carry it too; otherwise it collapses to a type default. This

/// matters for JOINs with `WITH TOTALS`: the default totals are produced

/// for the side that has no totals of its own (e.g. a constant subquery),

/// and which side that is can change depending on build/probe ordering

/// (`query_plan_join_swap_table`).

if (elem.column && isColumnConst(*elem.column))

{

columns.emplace_back(elem.column->cloneResized(1));

continue;

}

auto column = elem.type->createColumn();

column->insertDefault();

columns.emplace_back(std::move(column));

}

auto source = std::make_shared<SourceFromSingleChunk>(current_header, Chunk(std::move(columns), 1));

pipe.addTotalsSource(std::move(source));

{

checkInitializedAndNotCompleted();

/// It is possible that pipeline already have extremes.

/// For example, it may be added from VIEW subquery.

/// In this case, recalculate extremes again - they should be calculated for different rows.

if (pipe.getExtremesPort())

pipe.dropExtremes();

resize(1);

auto transform = std::make_shared<ExtremesTransform>(getSharedHeader());

auto * port = &transform->getExtremesPort();

pipe.addTransform(std::move(transform), nullptr, port);

VectorWithMemoryTracking<std::unique_ptr<QueryPipelineBuilder>> pipelines,

size_t max_threads_limit,

Processors * collected_processors)

{

if (pipelines.empty())

throw Exception(ErrorCodes::LOGICAL_ERROR, "Cannot unite an empty set of pipelines");

/// Should we limit the number of threads for united pipeline. True if all pipelines have max_threads != 0.

/// If true, result max_threads will be sum(max_threads).

/// Note: it may be > than settings.max_threads, so we should apply this limit again.

bool will_limit_max_threads = true;

size_t max_threads = 0;

bool concurrency_control = false;

Pipes pipes;

QueryPlanResourceHolder resources;

for (auto & pipeline_ptr : pipelines)

{

auto & pipeline = *pipeline_ptr;

pipeline.checkInitialized();

resources = std::move(pipeline.resources);

pipeline.pipe.collected_processors = collected_processors;

pipes.emplace_back(std::move(pipeline.pipe));

max_threads += pipeline.max_threads;

will_limit_max_threads = will_limit_max_threads && pipeline.max_threads != 0;

/// If one of pipelines uses more threads then current limit, will keep it.

/// It may happen if max_distributed_connections > max_threads

max_threads_limit = std::max(pipeline.max_threads, max_threads_limit);

// Use concurrency control if at least one of pipelines is using it

concurrency_control = concurrency_control || pipeline.getConcurrencyControl();

}

QueryPipelineBuilder pipeline;

pipeline.init(Pipe::unitePipes(std::move(pipes), collected_processors, false));

pipeline.addResources(resources);

if (will_limit_max_threads)

{

pipeline.setMaxThreads(max_threads);

pipeline.limitMaxThreads(max_threads_limit);

}

pipeline.setConcurrencyControl(concurrency_control);

pipeline.setCollectedProcessors(nullptr);

return pipeline;

QueryPipelineBuilderPtr left,

QueryPipelineBuilderPtr right,

ProcessorPtr transform,

Processors * collected_processors)

{

if (transform->getOutputs().size() != 1)

throw Exception(ErrorCodes::LOGICAL_ERROR, "Merge transform must have exactly 1 output, got {}", transform->getOutputs().size());

connect(*left->pipe.output_ports.front(), transform->getInputs().front());

connect(*right->pipe.output_ports.front(), transform->getInputs().back());

if (collected_processors)

collected_processors->emplace_back(transform);

left->pipe.output_ports.front() = &transform->getOutputs().front();

left->pipe.processors->emplace_back(transform);

left->pipe.processors->insert(left->pipe.processors->end(), right->pipe.processors->begin(), right->pipe.processors->end());

left->pipe.header = left->pipe.output_ports.front()->getSharedHeader();

left->pipe.max_parallel_streams = std::max(left->pipe.max_parallel_streams, right->pipe.max_parallel_streams);

left->resources = std::move(right->resources);

return left;

QueryPipelineBuilderPtr signal,

QueryPipelineBuilderPtr true_pipeline,

QueryPipelineBuilderPtr false_pipeline,

Processors * collected_processors)

{

if (!signal->getHeader().empty())

throw Exception(ErrorCodes::LOGICAL_ERROR, "Signal pipeline header must be empty in selectPipeline");

assertBlocksHaveEqualStructure(true_pipeline->getHeader(), false_pipeline->getHeader(), "selectPipeline");

size_t num_streams = std::max(true_pipeline->getNumStreams(), false_pipeline->getNumStreams());

signal->pipe.resize(1);

true_pipeline->pipe.resize(1);

false_pipeline->pipe.resize(1);

auto transform = std::make_shared<InputSelectorTransform>(true_pipeline->getHeader());

auto & transform_inputs = transform->getInputs();

auto it = transform_inputs.begin();

connect(*signal->pipe.output_ports.front(), *it++);

connect(*true_pipeline->pipe.output_ports.front(), *it++);

connect(*false_pipeline->pipe.output_ports.front(), *it);

if (collected_processors)

collected_processors->emplace_back(transform);

signal->pipe.output_ports.front() = &transform->getOutputs().front();

signal->pipe.processors->emplace_back(transform);

signal->pipe.processors->insert(signal->pipe.processors->end(), true_pipeline->pipe.processors->begin(), true_pipeline->pipe.processors->end());

signal->pipe.processors->insert(signal->pipe.processors->end(), false_pipeline->pipe.processors->begin(), false_pipeline->pipe.processors->end());

signal->pipe.header = signal->pipe.output_ports.front()->getSharedHeader();

signal->pipe.max_parallel_streams = std::max({signal->pipe.max_parallel_streams, true_pipeline->pipe.max_parallel_streams, false_pipeline->pipe.max_parallel_streams});

/// Transfer resources so they stay alive while processors execute.

signal->resources.append(true_pipeline->resources);

signal->resources.append(false_pipeline->resources);

signal->pipe.resize(num_streams);

return signal;

{

if (!processors)

return;

for (const auto & processor : *processors)

processor->setQueryPlanStep(join_step, static_cast<size_t>(stage));

std::unique_ptr<QueryPipelineBuilder> left,

std::unique_ptr<QueryPipelineBuilder> right,

JoinPtr join,

SharedHeader & out_header,

size_t max_block_size,

IQueryPlanStep * join_step,

Processors * collected_processors)

{

left->checkInitializedAndNotCompleted();

right->checkInitializedAndNotCompleted();

left->pipe.dropExtremes();

right->pipe.dropExtremes();

if ((left->getNumStreams() != 1 || right->getNumStreams() != 1) && join->getTableJoin().kind() == JoinKind::Paste)

{

left->pipe.resize(1);

right->pipe.resize(1);

}

else if (left->getNumStreams() != 1 || right->getNumStreams() != 1)

throw Exception(ErrorCodes::LOGICAL_ERROR, "Join is supported only for pipelines with one output port, got {} and {}", left->getNumStreams(), right->getNumStreams());

if (left->hasTotals() || right->hasTotals())

throw Exception(ErrorCodes::NOT_IMPLEMENTED, "Current join algorithm is supported only for pipelines without totals");

SharedHeaders inputs = {left->getSharedHeader(), right->getSharedHeader()};

if (join->getTableJoin().kind() == JoinKind::Paste)

{

auto joining = std::make_shared<PasteJoinTransform>(join, inputs, out_header, max_block_size);

auto result = mergePipelines(std::move(left), std::move(right), std::move(joining), collected_processors);

assignToJoinStage(collected_processors, join_step, JoinStep::JoinStage::Default);

return result;

}

auto joining = std::make_shared<MergeJoinTransform>(join, inputs, out_header, max_block_size);

auto result = mergePipelines(std::move(left), std::move(right), std::move(joining), collected_processors);

assignToJoinStage(collected_processors, join_step, JoinStep::JoinStage::Default);

return result;

std::unique_ptr<QueryPipelineBuilder> left,

std::unique_ptr<QueryPipelineBuilder> right,

ProcessorPtr joining,

Processors * collected_processors)

{

left->checkInitializedAndNotCompleted();

right->checkInitializedAndNotCompleted();

left->pipe.dropExtremes();

right->pipe.dropExtremes();

if (left->hasTotals() || right->hasTotals())

throw Exception(ErrorCodes::NOT_IMPLEMENTED, "Current join algorithm is supported only for pipelines without totals");

/// The joining transform may rely on the order of each input stream (e.g. IEJoin expects

/// pre-sorted inputs), so a multi-stream input cannot be silently squashed here.

if (left->getNumStreams() != 1 || right->getNumStreams() != 1)

throw Exception(ErrorCodes::LOGICAL_ERROR,

"Join is supported only for pipelines with one output port, got {} and {}", left->getNumStreams(), right->getNumStreams());

return mergePipelines(std::move(left), std::move(right), std::move(joining), collected_processors);

std::unique_ptr<QueryPipelineBuilder> left,

std::unique_ptr<QueryPipelineBuilder> right,

JoinPtr join,

SharedHeader & out_header,

size_t max_block_size,

IQueryPlanStep * join_step,

Processors * collected_processors)

{

left->checkInitializedAndNotCompleted();

right->checkInitializedAndNotCompleted();

left->pipe.dropExtremes();

right->pipe.dropExtremes();

if (left->getNumStreams() != right->getNumStreams())

throw Exception(ErrorCodes::LOGICAL_ERROR, "Join by layers is supported only for pipelines equal number of output ports, got {} and {}", left->getNumStreams(), right->getNumStreams());

if (left->hasTotals() || right->hasTotals())

throw Exception(ErrorCodes::NOT_IMPLEMENTED, "Current join algorithm is supported only for pipelines without totals");

SharedHeaders inputs = {left->getSharedHeader(), right->getSharedHeader()};

size_t num_streams = left->getNumStreams();

for (size_t i = 0; i < num_streams; ++i)

{

auto joining = std::make_shared<MergeJoinTransform>(join->clone(std::make_shared<TableJoin>(join->getTableJoin()), left->getSharedHeader(), right->getSharedHeader()), inputs, out_header, max_block_size);

connect(*left->pipe.output_ports[i], joining->getInputs().front());

connect(*right->pipe.output_ports[i], joining->getInputs().back());

left->pipe.output_ports[i] = &joining->getOutputPort();

if (collected_processors)

collected_processors->emplace_back(joining);

left->pipe.processors->emplace_back(std::move(joining));

}

left->pipe.processors->insert(left->pipe.processors->end(), right->pipe.processors->begin(), right->pipe.processors->end());

left->pipe.header = left->pipe.output_ports.front()->getSharedHeader();

left->pipe.max_parallel_streams = std::max(left->pipe.max_parallel_streams, right->pipe.max_parallel_streams);

left->resources = std::move(right->resources);

assignToJoinStage(collected_processors, join_step, JoinStep::JoinStage::Default);

return left;

std::unique_ptr<QueryPipelineBuilder> left,

std::unique_ptr<QueryPipelineBuilder> right,

JoinPtr join,

SharedHeader & output_header,

size_t max_block_size,

size_t min_block_size_rows,

size_t min_block_size_bytes,

size_t max_streams,

IQueryPlanStep * join_step,

bool keep_left_read_in_order,

Processors * collected_processors)

{

left->checkInitializedAndNotCompleted();

right->checkInitializedAndNotCompleted();

/// Extremes before join are useless. They will be calculated after if needed.

left->pipe.dropExtremes();

right->pipe.dropExtremes();

left->pipe.collected_processors = collected_processors;

auto add_to_left_pipe = [&](ProcessorPtr processor)

{

if (collected_processors)

collected_processors->emplace_back(processor);

left->pipe.processors->emplace_back(std::move(processor));

};

/// Collect the NEW processors for the right pipeline.

QueryPipelineProcessorsCollector collector(*right, join_step);

/// In case joined subquery has totals, and we don't, add default chunk to totals.

bool default_totals = false;

if (!left->hasTotals() && right->hasTotals())

{

left->addDefaultTotals();

default_totals = true;

}

/// (left) ──────┐

/// ╞> Joining ─> (joined)

/// (left) ─┐┌───┘

/// └┼───┐

/// (right) ┐ (totals) ──┼─┐ ╞> Joining ─> (joined)

/// ╞> Resize ┐ ╓─┘┌┼─┘

/// (right) ┘ │ ╟──┘└─┐

/// ╞> FillingJoin ─> Resize ╣ ╞> Joining ─> (totals)

/// (totals) ─────────┘ ╙─────┘

size_t num_streams = left->getNumStreams();

if (join->supportParallelJoin() && !right->hasTotals())

{

if (!keep_left_read_in_order)

{

left->resize(max_streams);

num_streams = max_streams;

}

auto filling_finish_counter = std::make_shared<FinishCounter>(max_streams);

right->resize(max_streams);

auto concurrent_right_filling_transform = [&](const OutputPortRawPtrs & outports)

{

Processors processors;

if (min_block_size_rows > 0 || min_block_size_bytes > 0)

{

for (const auto & outport : outports)

{

auto squashing = std::make_shared<SimpleSquashingChunksTransform>(right->getSharedHeader(), min_block_size_rows, min_block_size_bytes);

connect(*outport, squashing->getInputs().front());

processors.emplace_back(squashing);

auto adding_joined = std::make_shared<FillingRightJoinSideTransform>(right->getSharedHeader(), join, filling_finish_counter);

connect(squashing->getOutputPort(), adding_joined->getInputs().front());

processors.emplace_back(std::move(adding_joined));

}

}

else

{

for (const auto & outport : outports)

{

auto adding_joined = std::make_shared<FillingRightJoinSideTransform>(right->getSharedHeader(), join, filling_finish_counter);

connect(*outport, adding_joined->getInputs().front());

processors.emplace_back(std::move(adding_joined));

}

}

return processors;

};

right->transform(concurrent_right_filling_transform);

right->resize(1);

}

else

{

right->resize(1);

auto filling_finish_counter = std::make_shared<FinishCounter>(1);

auto adding_joined = std::make_shared<FillingRightJoinSideTransform>(right->getSharedHeader(), join, filling_finish_counter);

InputPort * totals_port = nullptr;

if (right->hasTotals())

totals_port = adding_joined->addTotalsPort();

right->addTransform(std::move(adding_joined), totals_port, nullptr);

}

size_t num_streams_including_totals = num_streams + (left->hasTotals() ? 1 : 0);

right->resize(num_streams_including_totals);

auto lit = left->pipe.output_ports.begin();

auto rit = right->pipe.output_ports.begin();

OutputPortRawPtrs joined_output_ports;

OutputPortRawPtrs delayed_root_output_ports;

OutputPortRawPtrs non_joined_output_ports;

std::shared_ptr<DelayedJoinedBlocksTransform> delayed_root = nullptr;

if (join->hasDelayedBlocks())

{

delayed_root = std::make_shared<DelayedJoinedBlocksTransform>(num_streams, join);

if (!delayed_root->getInputs().empty() || delayed_root->getOutputs().size() != num_streams)

throw Exception(ErrorCodes::LOGICAL_ERROR,

"DelayedJoinedBlocksTransform should have no inputs and {} outputs, "

"but has {} inputs and {} outputs",

num_streams, delayed_root->getInputs().size(), delayed_root->getOutputs().size());

add_to_left_pipe(delayed_root);

for (auto & outport : delayed_root->getOutputs())

delayed_root_output_ports.emplace_back(&outport);

}

/// When true, non-joined rows are emitted by separate NonJoinedBlocksTransform sources in parallel.

/// Can be combined with delayed_root: in that case two DelayedPorts are created:

/// - the first for the DelayedJoinedBlocksWorkerTransforms until all JoiningTransforms finish

/// - the second for the NonJoinedBlocksTransforms until the first DelayedPorts finishes

bool use_parallel_non_joined = join->supportParallelNonJoinedBlocksProcessing();

/// When delayed blocks exist, JoiningTransform still needs a finish_counter so it can emit

/// non-joined rows for the main bucket (GraceHashJoin spilled case).

/// When only parallel non-joined (no delayed blocks), nullptr disables non-joined emission

/// inside JoiningTransform entirely (the parallel NonJoinedBlocksTransform handles it).

auto joining_finish_counter = (use_parallel_non_joined && !delayed_root)

? nullptr

: std::make_shared<FinishCounter>(num_streams);

SharedHeader left_header = left->getSharedHeader();

for (size_t i = 0; i < num_streams; ++i)

{

OutputPort * left_port = *lit;

if (min_block_size_rows > 0 || min_block_size_bytes > 0)

{

auto squashing = std::make_shared<SimpleSquashingChunksTransform>(left->getSharedHeader(), min_block_size_rows, min_block_size_bytes);

connect(*left_port, squashing->getInputs().front());

left_port = &squashing->getOutputPort();

add_to_left_pipe(std::move(squashing));

}

auto joining = std::make_shared<JoiningTransform>(

left_header, output_header, join, max_block_size, false, default_totals, joining_finish_counter);

connect(*left_port, joining->getInputs().front());

connect(**rit, joining->getInputs().back());

if (delayed_root)

{

// Process delayed joined blocks when all JoiningTransform are finished.

auto delayed = std::make_shared<DelayedJoinedBlocksWorkerTransform>(

output_header,

[left_header, output_header, max_block_size, join]()

{ return join->getNonJoinedBlocks(*left_header, *output_header, max_block_size); });

if (delayed->getInputs().size() != 1 || delayed->getOutputs().size() != 1)

throw Exception(ErrorCodes::LOGICAL_ERROR, "DelayedJoinedBlocksWorkerTransform should have one input and one output");

connect(*delayed_root_output_ports[i], delayed->getInputs().front());

joined_output_ports.push_back(&joining->getOutputs().front());

joined_output_ports.push_back(&delayed->getOutputs().front());

add_to_left_pipe(std::move(delayed));

if (use_parallel_non_joined)

{

auto non_joined = std::make_shared<NonJoinedBlocksTransform>(

output_header, join, *left_header, max_block_size, i, num_streams);

non_joined_output_ports.push_back(&non_joined->getPort());

add_to_left_pipe(std::move(non_joined));

}

}

else if (use_parallel_non_joined)

{

/// Parallel non-joined source, activated by DelayedPortsProcessor after all probing finishes

auto non_joined = std::make_shared<NonJoinedBlocksTransform>(

output_header, join, *left_header, max_block_size, i, num_streams);

joined_output_ports.push_back(&joining->getOutputs().front());

joined_output_ports.push_back(&non_joined->getPort());

add_to_left_pipe(std::move(non_joined));

}

else

{

*lit = &joining->getOutputs().front();

}

++lit;

++rit;

add_to_left_pipe(std::move(joining));

}

if (delayed_root || use_parallel_non_joined)

{

// Process DelayedJoinedBlocksTransform after all JoiningTransforms.

VectorWithMemoryTracking<UInt64> delayed_ports_numbers;

delayed_ports_numbers.reserve(joined_output_ports.size() / 2);

for (size_t i = 1; i < joined_output_ports.size(); i += 2)

delayed_ports_numbers.push_back(i);

auto delayed_processor = std::make_shared<DelayedPortsProcessor>(output_header, 2 * num_streams, delayed_ports_numbers);

add_to_left_pipe(delayed_processor);

// Connect @delayed_processor ports with inputs (JoiningTransforms & DelayedJoinedBlocksTransforms) / pipe outputs

auto next_delayed_input = delayed_processor->getInputs().begin();

for (OutputPort * port : joined_output_ports)

connect(*port, *next_delayed_input++);

left->pipe.output_ports.clear();

for (OutputPort & port : delayed_processor->getOutputs())

left->pipe.output_ports.push_back(&port);

left->pipe.header = output_header;

left->resize(num_streams);

// Second DelayedPortsProcessor: when both delayed blocks and parallel non-joined are active,

// delays the NonJoinedBlocksTransforms until all delayed-block workers finish.

if (delayed_root && use_parallel_non_joined)

{

VectorWithMemoryTracking<UInt64> non_joined_delayed_ports;

non_joined_delayed_ports.reserve(num_streams);

for (size_t i = 0; i < num_streams; ++i)

non_joined_delayed_ports.push_back(2 * i + 1);

auto non_joined_processor = std::make_shared<DelayedPortsProcessor>(

output_header, 2 * num_streams, non_joined_delayed_ports);

add_to_left_pipe(non_joined_processor);

auto next_input = non_joined_processor->getInputs().begin();

for (size_t i = 0; i < num_streams; ++i)

{

connect(*left->pipe.output_ports[i], *next_input++);

connect(*non_joined_output_ports[i], *next_input++);

}

left->pipe.output_ports.clear();

for (OutputPort & port : non_joined_processor->getOutputs())

left->pipe.output_ports.push_back(&port);

left->pipe.header = output_header;

left->resize(num_streams);

}

}

if (left->hasTotals())

{

auto joining = std::make_shared<JoiningTransform>(left_header, output_header, join, max_block_size, true, default_totals);

connect(*left->pipe.totals_port, joining->getInputs().front());

connect(**rit, joining->getInputs().back());

left->pipe.totals_port = &joining->getOutputs().front();

++rit;

add_to_left_pipe(std::move(joining));

}

assignToJoinStage(collected_processors, join_step, JoinStep::JoinStage::Probe);

/// Move the collected processors to the last step in the right pipeline.

Processors processors = collector.detachProcessors(static_cast<size_t>(JoinStep::JoinStage::Build));

if (join_step)

join_step->appendExtraProcessors(processors);

left->pipe.processors->insert(left->pipe.processors->end(), right->pipe.processors->begin(), right->pipe.processors->end());

left->resources = std::move(right->resources);

left->pipe.header = left->pipe.output_ports.front()->getSharedHeader();

left->pipe.max_parallel_streams = std::max(left->pipe.max_parallel_streams, right->pipe.max_parallel_streams);

return left;

std::unique_ptr<QueryPipelineBuilder> left,

std::unique_ptr<QueryPipelineBuilder> right,

JoinPtr join,

SharedHeader & output_header,

size_t max_block_size,

IQueryPlanStep * join_step,

Processors * collected_processors)

{

left->checkInitializedAndNotCompleted();

right->checkInitializedAndNotCompleted();

/// Extremes before join are useless. They will be calculated after if needed.

left->pipe.dropExtremes();

right->pipe.dropExtremes();

left->pipe.collected_processors = collected_processors;

/// Collect the NEW processors for the right pipeline.

QueryPipelineProcessorsCollector collector(*right, join_step);

if (left->hasTotals() || right->hasTotals())

throw Exception(ErrorCodes::NOT_IMPLEMENTED, "Join by layers is supported only for pipelines without totals");

if (!join->isCloneSupported())

throw Exception(ErrorCodes::LOGICAL_ERROR, "Clone is not supported for {}", join->getName());

if (join->hasDelayedBlocks())

throw Exception(ErrorCodes::LOGICAL_ERROR, "Delayed ports from {} are not supported with join by layers", join->getName());

/// (left) ──────┐

/// ╞> Joining ─> (joined)

/// (left) ─┐┌───┘

/// └┼───┐

/// (right) ─> FillingJoin ──────────────────────┘ ╞> Joining ─> (joined)

/// (right) ─> FillingJoin ──────────────────────────┘

size_t num_streams = left->getNumStreams();

if (right->getNumStreams() != num_streams)

throw Exception(ErrorCodes::LOGICAL_ERROR,

"Join by layers expects the same number of ports for the left and the right pipeline. Actual: {} and {}.",

num_streams, right->getNumStreams());

SharedHeader left_header = left->getSharedHeader();

VectorWithMemoryTracking<JoinPtr> joins;

right->addSimpleTransform([&](const SharedHeader & header)

{

joins.push_back(join->cloneNoParallel(std::make_shared<TableJoin>(join->getTableJoin()), left->getSharedHeader(), header));

auto finish_counter = std::make_shared<FinishCounter>(1);

return std::make_shared<FillingRightJoinSideTransform>(header, joins.back(), finish_counter);

});

auto lit = left->pipe.output_ports.begin();

auto rit = right->pipe.output_ports.begin();

for (size_t i = 0; i < num_streams; ++i)

{

auto finish_counter = std::make_shared<FinishCounter>(1);

auto joining = std::make_shared<JoiningTransform>(

left_header, output_header, joins[i], max_block_size, false, false, finish_counter);

connect(**lit, joining->getInputs().front());

connect(**rit, joining->getInputs().back());

*lit = &joining->getOutputs().front();

++lit;

++rit;

if (collected_processors)

collected_processors->emplace_back(joining);

left->pipe.processors->emplace_back(std::move(joining));

}

assignToJoinStage(collected_processors, join_step, JoinStep::JoinStage::Probe);

/// Move the collected processors to the last step in the right pipeline.

Processors processors = collector.detachProcessors(static_cast<size_t>(JoinStep::JoinStage::Build));

if (join_step)

join_step->appendExtraProcessors(processors);

left->pipe.processors->insert(left->pipe.processors->end(), right->pipe.processors->begin(), right->pipe.processors->end());

left->resources = std::move(right->resources);

left->pipe.header = left->pipe.output_ports.front()->getSharedHeader();

left->pipe.max_parallel_streams = std::max(left->pipe.max_parallel_streams, right->pipe.max_parallel_streams);

return left;

SharedHeader res_header,

SetAndKeyPtr set_and_key,

const SizeLimits & limits,

PreparedSetsCachePtr prepared_sets_cache)

{

dropTotalsAndExtremes();

resize(1);

auto transform = std::make_shared<CreatingSetsTransform>(

getSharedHeader(),

res_header,

std::move(set_and_key),

limits,

std::move(prepared_sets_cache));

pipe.addTransform(std::move(transform));

SharedHeader res_header,

MaterializedCTEPtr materialized_cte

)

{

checkInitializedAndNotCompleted();

dropTotalsAndExtremes();

resize(1);

auto transform = std::make_shared<MaterializingCTETransform>(

getSharedHeader(),

res_header,

std::move(materialized_cte));

pipe.addTransform(std::move(transform));

{

checkInitializedAndNotCompleted();

if (!pipeline.getHeader().empty())

throw Exception(ErrorCodes::LOGICAL_ERROR, "Pipeline for CreatingSets should have empty header. Got: {}",

pipeline.getHeader().dumpStructure());

pipeline.dropTotalsAndExtremes();

bool has_totals = pipe.getTotalsPort();

bool has_extremes = pipe.getExtremesPort();

size_t num_extra_ports = (has_totals ? 1 : 0) + (has_extremes ? 1 : 0);

VectorWithMemoryTracking<UInt64> delayed_streams(pipe.numOutputPorts() + num_extra_ports);

iota(delayed_streams.data(), delayed_streams.size(), UInt64{0});

auto * collected_processors = pipe.collected_processors;

Pipes pipes;

pipes.emplace_back(std::move(pipe));

pipes.emplace_back(QueryPipelineBuilder::getPipe(std::move(pipeline), resources));

pipe = Pipe::unitePipes(std::move(pipes), collected_processors, true);

auto processor = std::make_shared<DelayedPortsProcessor>(getSharedHeader(), pipe.numOutputPorts() + num_extra_ports, delayed_streams, true);

auto in = processor->getInputs().begin();

auto out = processor->getOutputs().begin();

InputPort * totals_in = has_totals ? &*(in++) : nullptr;

InputPort * extremes_in = has_extremes ? &*(in++) : nullptr;

OutputPort * totals_out = has_totals ? &*(out++) : nullptr;

OutputPort * extremes_out = has_extremes ? &*(out++) : nullptr;

pipe.addTransform(std::move(processor), totals_in, extremes_in, totals_out, extremes_out);

{

if (!isCompleted())

throw Exception(ErrorCodes::LOGICAL_ERROR, "Cannot execute pipeline because it is not completed");

return std::make_shared<PipelineExecutor>(pipe.processors, process_list_element);

{

resources = std::move(pipeline.resources);

return std::move(pipeline.pipe);

{

QueryPipeline res(std::move(builder.pipe));

res.addResources(std::move(builder.resources));

res.setNumThreads(builder.getNumThreads());

res.setConcurrencyControl(builder.getConcurrencyControl());

res.setProcessListElement(builder.process_list_element);

res.setProgressCallback(builder.progress_callback);

return res;

: pipeline(pipeline_), step(step_)

{

pipeline.setCollectedProcessors(&processors);

{

for (auto & processor : processors)

processor->setQueryPlanStep(step, group);

Processors res;

res.swap(processors);

return res;