{

if (!input.isConnected())

throw Exception(

ErrorCodes::LOGICAL_ERROR,

"Cannot create QueryPipeline because {} has disconnected input",

processor->getName());

{

if (!output.isConnected())

{

WriteBufferFromOwnString out;

if (!processors.empty())

printPipeline(processors, out);

throw Exception(

ErrorCodes::LOGICAL_ERROR,

"Cannot create QueryPipeline because {} {} has disconnected output: {}",

processor->getName(), processor->getDescription(), out.str());

}

Processors & processors,

OutputPort * output,

OutputPort * totals,

OutputPort * extremes)

{

if (!output || output->isConnected())

throw Exception(

ErrorCodes::LOGICAL_ERROR,

"Cannot create pulling QueryPipeline because its output port is connected or null");

if (totals && totals->isConnected())

throw Exception(

ErrorCodes::LOGICAL_ERROR,

"Cannot create pulling QueryPipeline because its totals port is connected");

if (extremes && extremes->isConnected())

throw Exception(

ErrorCodes::LOGICAL_ERROR,

"Cannot create pulling QueryPipeline because its extremes port is connected");

bool found_output = false;

bool found_totals = false;

bool found_extremes = false;

for (const auto & processor : processors)

{

for (const auto & in : processor->getInputs())

checkInput(in, processor);

for (const auto & out : processor->getOutputs())

{

if (&out == output)

found_output = true;

else if (totals && &out == totals)

found_totals = true;

else if (extremes && &out == extremes)

found_extremes = true;

else

checkOutput(out, processor, processors);

}

}

if (!found_output)

throw Exception(

ErrorCodes::LOGICAL_ERROR,

"Cannot create pulling QueryPipeline because its output port does not belong to any processor");

if (totals && !found_totals)

throw Exception(

ErrorCodes::LOGICAL_ERROR,

"Cannot create pulling QueryPipeline because its totals port does not belong to any processor");

if (extremes && !found_extremes)

throw Exception(

ErrorCodes::LOGICAL_ERROR,

"Cannot create pulling QueryPipeline because its extremes port does not belong to any processor");

{

for (const auto & processor : processors)

{

for (const auto & in : processor->getInputs())

checkInput(in, processor);

for (const auto & out : processor->getOutputs())

checkOutput(out, processor);

}

{

RowsBeforeStepCounterPtr rows_before_limit_at_least;

UnorderedSetWithMemoryTracking<IProcessor *> processors;

/// Start at the output and follow inputs toward the sources. For each path, remember which

/// limit is being counted and which of its input ports the path came through. A shared processor

/// may need to be visited more than once when it feeds different limits or limit inputs.

///

/// `counted_inputs_by_limit` records limit inputs whose rows are counted closer to the source.

/// The limit itself counts rows only for the other inputs.

MapWithMemoryTracking<IProcessor *, UnorderedSetWithMemoryTracking<size_t>> counted_inputs_by_limit;

MapWithMemoryTracking<std::tuple<IProcessor *, IProcessor *, ssize_t>, bool> visited;

bool has_limit = false;

struct QueuedEntry

{

IProcessor * processor;

IProcessor * limit_being_counted;

ssize_t limit_input_port;

};

QueueWithMemoryTracking<QueuedEntry> queue;

auto mark_limit_input_as_counted = [&](IProcessor * limit, ssize_t input_port)

{

if (input_port < 0 || static_cast<size_t>(input_port) >= limit->getInputs().size())

throw Exception(

ErrorCodes::LOGICAL_ERROR,

"Invalid input port {} while placing the rows_before_limit_at_least counter",

input_port);

counted_inputs_by_limit[limit].emplace(static_cast<size_t>(input_port));

};

queue.push({ output_format, nullptr, -1 });

while (!queue.empty())

{

auto * processor = queue.front().processor;

auto * limit_being_counted = queue.front().limit_being_counted;

auto limit_input_port = queue.front().limit_input_port;

queue.pop();

/// The same processor can be reached under different limits or parent input ports.

if (!visited.emplace(std::tuple(processor, limit_being_counted, limit_input_port), true).second)

continue;

/// Set counter based on the following cases:

/// 1. Remote: Set counter on Remote

/// 2. Limit ... PartialSorting: Set counter on PartialSorting

/// 3. Limit ... TotalsHaving(with filter) ... Remote: Set counter on the input port of Limit

/// 4. Limit ... MergingAggregated ... Remote: Set counter on the input port of Limit

/// 5. Limit ... Remote: Set counter on Remote

/// 6. Limit ... LimitBy: Set counter on LimitBy, as it may not be executed on initiator

/// 7. Limit ... : Set counter on the input port of Limit

/// Case 1.

if ((typeid_cast<RemoteSource *>(processor) || typeid_cast<DelayedSource *>(processor)) && !limit_being_counted)

{

processors.emplace(processor);

continue;

}

auto * limit = typeid_cast<LimitTransform *>(processor);

auto * negative_limit = typeid_cast<NegativeLimitTransform *>(processor);

if (((limit && limit->isShardLimit()) || (negative_limit && negative_limit->isShardLimit()))

&& limit_being_counted)

{

/// Rows discarded by a shard limit still belong to the parent limit's total. Mark the

/// parent input as counted, then continue toward the source past the shard limit.

mark_limit_input_as_counted(limit_being_counted, limit_input_port);

limit_being_counted = processor;

counted_inputs_by_limit.try_emplace(limit_being_counted);

}

else if (limit || negative_limit || typeid_cast<FractionalLimitTransform *>(processor))

{

has_limit = true;

/// A limit from the query changes the rows seen by an outer limit. Do not count through it.

if (limit_being_counted)

continue;

limit_being_counted = processor;

counted_inputs_by_limit.try_emplace(limit_being_counted);

}

else if (limit_being_counted)

{

/// Case 2.

if (typeid_cast<PartialSortingTransform *>(processor))

{

processors.emplace(processor);

mark_limit_input_as_counted(limit_being_counted, limit_input_port);

continue;

}

/// Case 3.

if (auto * having = typeid_cast<TotalsHavingTransform *>(processor))

{

if (having->hasFilter())

continue;

}

/// Case 4.

if (typeid_cast<MergingAggregatedTransform *>(processor) || typeid_cast<MergingAggregatedBucketTransform *>(processor)

|| typeid_cast<SortingAggregatedTransform *>(processor)

|| typeid_cast<SortingAggregatedForMemoryBoundMergingTransform *>(processor))

{

continue;

}

/// Case 5.

if (typeid_cast<RemoteSource *>(processor) || typeid_cast<DelayedSource *>(processor))

{

processors.emplace(processor);

mark_limit_input_as_counted(limit_being_counted, limit_input_port);

continue;

}

/// Case 6.

if (typeid_cast<LimitByTransform *>(processor) || typeid_cast<LimitBySortedStreamTransform *>(processor))

{

processors.emplace(processor);

mark_limit_input_as_counted(limit_being_counted, limit_input_port);

continue;

}

}

/// Skip totals and extremes port for output format.

if (auto * format = dynamic_cast<IOutputFormat *>(processor))

{

auto * child_processor = &format->getPort(IOutputFormat::PortKind::Main).getOutputPort().getProcessor();

queue.push({ child_processor, limit_being_counted, limit_input_port });

continue;

}

/// Skip CreatingSetsTransform

if (typeid_cast<CreatingSetsTransform *>(processor))

continue;

if (limit_being_counted == processor)

{

ssize_t i = 0;

for (auto & child_port : processor->getInputs())

{

auto * child_processor = &child_port.getOutputPort().getProcessor();

queue.push({ child_processor, limit_being_counted, i });

++i;

}

}

else

{

for (auto & child_port : processor->getInputs())

{

auto * child_processor = &child_port.getOutputPort().getProcessor();

queue.push({ child_processor, limit_being_counted, limit_input_port });

}

}

}

/// Case 7.

for (auto && [limit, ports] : counted_inputs_by_limit)

{

/// If there are some input ports which don't have the counter, add it to the limit processor.

if (ports.size() < limit->getInputs().size())

{

processors.emplace(limit);

for (auto port : ports)

{

if (auto * lim = typeid_cast<LimitTransform *>(limit))

lim->setInputPortHasCounter(port);

else if (auto * neg_lim = typeid_cast<NegativeLimitTransform *>(limit))

neg_lim->setInputPortHasCounter(port);

else if (auto * frac_lim = typeid_cast<FractionalLimitTransform *>(limit))

frac_lim->setInputPortHasCounter(port);

}

}

}

if (!processors.empty())

{

rows_before_limit_at_least = std::make_shared<RowsBeforeStepCounter>();

for (const auto & processor : processors)

processor->setRowsBeforeLimitCounter(rows_before_limit_at_least);

/// If there is a limit, then enable rows_before_limit_at_least

/// It is needed when zero rows is read, but we still want rows_before_limit_at_least in result.

if (has_limit)

rows_before_limit_at_least->add(0);

output_format->setRowsBeforeLimitCounter(rows_before_limit_at_least);

}

{

bool has_aggregation = false;

if (!processors->empty())

{

RowsBeforeStepCounterPtr rows_before_aggregation = std::make_shared<RowsBeforeStepCounter>();

for (const auto & processor : *processors)

{

if (typeid_cast<AggregatingTransform *>(processor.get()) || typeid_cast<AggregatingInOrderTransform *>(processor.get()))

{

processor->setRowsBeforeAggregationCounter(rows_before_aggregation);

has_aggregation = true;

}

if (typeid_cast<RemoteSource *>(processor.get()) || typeid_cast<DelayedSource *>(processor.get()))

processor->setRowsBeforeAggregationCounter(rows_before_aggregation);

}

if (has_aggregation)

rows_before_aggregation->add(0);

output_format->setRowsBeforeAggregationCounter(rows_before_aggregation);

}

QueryPlanResourceHolder resources_,

std::shared_ptr<Processors> processors_)

: resources(std::move(resources_))

, processors(std::move(processors_))

{

checkCompleted(*processors);

QueryPlanResourceHolder resources_,

std::shared_ptr<Processors> processors_,

InputPort * input_)

: resources(std::move(resources_))

, processors(std::move(processors_))

, input(input_)

{

if (!input || input->isConnected())

throw Exception(

ErrorCodes::LOGICAL_ERROR,

"Cannot create pushing QueryPipeline because its input port is connected or null");

bool found_input = false;

for (const auto & processor : *processors)

{

for (const auto & in : processor->getInputs())

{

if (&in == input)

found_input = true;

else

checkInput(in, processor);

}

for (const auto & out : processor->getOutputs())

checkOutput(out, processor);

}

if (!found_input)

throw Exception(

ErrorCodes::LOGICAL_ERROR,

"Cannot create pushing QueryPipeline because its input port does not belong to any processor");

QueryPlanResourceHolder resources_,

std::shared_ptr<Processors> processors_,

OutputPort * output_,

OutputPort * totals_,

OutputPort * extremes_)

: resources(std::move(resources_))

, processors(std::move(processors_))

, output(output_)

, totals(totals_)

, extremes(extremes_)

{

checkPulling(*processors, output, totals, extremes);

{

if (pipe.numOutputPorts() > 0)

{

pipe.resize(1);

output = pipe.getOutputPort(0);

totals = pipe.getTotalsPort();

extremes = pipe.getExtremesPort();

processors = std::move(pipe.processors);

checkPulling(*processors, output, totals, extremes);

}

else

{

processors = std::move(pipe.processors);

checkCompleted(*processors);

}

: resources(chain.detachResources())

, processors(std::make_shared<Processors>())

, input(&chain.getInputPort())

, num_threads(chain.getNumThreads())

{

for (auto processor : chain.getProcessors())

processors->emplace_back(std::move(processor));

auto sink = std::make_shared<EmptySink>(chain.getOutputPort().getSharedHeader());

connect(chain.getOutputPort(), sink->getPort());

processors->emplace_back(std::move(sink));

input = &chain.getInputPort();

: processors(std::make_shared<Processors>())

{

auto & format_main = format->getPort(IOutputFormat::PortKind::Main);

auto & format_totals = format->getPort(IOutputFormat::PortKind::Totals);

auto & format_extremes = format->getPort(IOutputFormat::PortKind::Extremes);

if (!totals)

{

auto source = std::make_shared<NullSource>(format_totals.getSharedHeader());

totals = &source->getPort();

processors->emplace_back(std::move(source));

}

if (!extremes)

{

auto source = std::make_shared<NullSource>(format_extremes.getSharedHeader());

extremes = &source->getPort();

processors->emplace_back(std::move(source));

}

connect(*totals, format_totals);

connect(*extremes, format_extremes);

input = &format_main;

totals = nullptr;

extremes = nullptr;

output_format = format.get();

processors->emplace_back(std::move(format));

OutputPort *& output, OutputPort *& totals, OutputPort *& extremes, Processors & processors)

{

if (!totals && !extremes)

return;

chassert(output);

auto dropping = std::make_shared<DroppingTransform>(

output->getSharedHeader(),

/*num_streams_=*/1,

totals ? totals->getSharedHeader() : nullptr,

extremes ? extremes->getSharedHeader() : nullptr);

connect(*output, dropping->getInputs().front());

if (totals)

{

connect(*totals, *dropping->getTotalsPort());

totals = nullptr;

}

if (extremes)

{

connect(*extremes, *dropping->getExtremesPort());

extremes = nullptr;

}

output = &dropping->getOutputs().front();

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

{

if (!pulling())

throw Exception(ErrorCodes::LOGICAL_ERROR, "Pipeline must be pulling to be completed with sink");

dropTotalsAndExtremesViaTransform(output, totals, extremes, *processors);

connect(*output, sink->getPort());

processors->emplace_back(std::move(sink));

output = nullptr;

{

if (!pulling())

throw Exception(ErrorCodes::LOGICAL_ERROR, "Pipeline must be pulling to be completed with chain");

resources = chain.detachResources();

dropTotalsAndExtremesViaTransform(output, totals, extremes, *processors);

for (auto processor : chain.getProcessors())

processors->emplace_back(std::move(processor));

auto sink = std::make_shared<EmptySink>(chain.getOutputPort().getSharedHeader());

connect(*output, chain.getInputPort());

connect(chain.getOutputPort(), sink->getPort());

processors->emplace_back(std::move(sink));

output = nullptr;

{

if (!pushing())

throw Exception(ErrorCodes::LOGICAL_ERROR, "Pipeline must be pushing to be completed with pipe");

pipe.resize(1);

pipe.dropTotalsAndExtremes();

connect(*pipe.getOutputPort(0), *input);

input = nullptr;

auto pipe_processors = Pipe::detachProcessors(std::move(pipe));

processors->insert(processors->end(), pipe_processors.begin(), pipe_processors.end());

{

if (!output)

return;

auto materializing = std::make_shared<MaterializingTransform>(output->getSharedHeader(), remove_special_column_representations);

connect(*output, materializing->getInputPort());

output = &materializing->getOutputPort();

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

{

if (!pulling())

throw Exception(ErrorCodes::LOGICAL_ERROR, "Pipeline must be pulling to be completed with output format");

if (format->expectMaterializedColumns())

{

bool remove_special_column_representations = !format->supportsSpecialSerializationKinds();

addMaterializing(output, *processors, remove_special_column_representations);

addMaterializing(totals, *processors, remove_special_column_representations);

addMaterializing(extremes, *processors, remove_special_column_representations);

}

auto & format_main = format->getPort(IOutputFormat::PortKind::Main);

auto & format_totals = format->getPort(IOutputFormat::PortKind::Totals);

auto & format_extremes = format->getPort(IOutputFormat::PortKind::Extremes);

if (!totals)

{

auto source = std::make_shared<NullSource>(format_totals.getSharedHeader());

totals = &source->getPort();

processors->emplace_back(std::move(source));

}

if (!extremes)

{

auto source = std::make_shared<NullSource>(format_extremes.getSharedHeader());

extremes = &source->getPort();

processors->emplace_back(std::move(source));

}

connect(*output, format_main);

connect(*totals, format_totals);

connect(*extremes, format_extremes);

output = nullptr;

totals = nullptr;

extremes = nullptr;

initRowsBeforeLimit(format.get());

for (const auto & context : resources.interpreter_context)

{

if (context->getSettingsRef()[Setting::rows_before_aggregation])

{

initRowsBeforeAggregation(processors, format.get());

break;

}

}

output_format = format.get();

processors->emplace_back(std::move(format));

{

if (input)

return input->getHeader();

if (output)

return output->getHeader();

throw Exception(ErrorCodes::LOGICAL_ERROR, "Header is available only for pushing or pulling QueryPipeline");

{

if (input)

return input->getSharedHeader();

if (output)

return output->getSharedHeader();

throw Exception(ErrorCodes::LOGICAL_ERROR, "Header is available only for pushing or pulling QueryPipeline");

{

process_list_element = elem;

if (pushing())

{

if (auto * counting = dynamic_cast<CountingTransform *>(&input->getProcessor()))

{

counting->setProcessListElement(elem);

}

}

{

if (!pulling())

throw Exception(

ErrorCodes::LOGICAL_ERROR,

"It is possible to set limits and quota only to pulling QueryPipeline");

auto transform = std::make_shared<LimitsCheckingTransform>(output->getSharedHeader(), limits);

transform->setQuota(quota_);

transform->setNormalizedQueryHash(normalized_query_hash);

connect(*output, transform->getInputPort());

output = &transform->getOutputPort();

processors->emplace_back(std::move(transform));

{

if (!output_format)

return false;

result_rows = output_format->getResultRows();

result_bytes = output_format->getResultBytes();

return true;

{

chassert(pulling());

/// Attach a special transform to all output ports (result + possibly totals/extremes). The only purpose of the transform is to write

/// each chunk into the query result cache. All transforms hold a refcounted reference to the same query result cache writer object.

/// This ensures that all transforms write to the single same cache entry. The writer object synchronizes internally, the expensive

/// stuff like cloning chunks happens outside lock scopes).

auto add_stream_in_query_result_cache_transform = [&](OutputPort *& out_port, QueryResultCacheWriter::ChunkType chunk_type)

{

if (!out_port)

return;

auto transform = std::make_shared<StreamInQueryResultCacheTransform>(out_port->getHeader(), query_result_cache_writer, chunk_type);

connect(*out_port, transform->getInputPort());

out_port = &transform->getOutputPort();

processors->emplace_back(std::move(transform));

};

using enum QueryResultCacheWriter::ChunkType;

add_stream_in_query_result_cache_transform(output, Result);

add_stream_in_query_result_cache_transform(totals, Totals);

add_stream_in_query_result_cache_transform(extremes, Extremes);

{

/// QueryPipeline can contain multiple StreamInQueryResultCacheTransforms,

/// and all StreamInQueryResultCacheTransforms can point to different QueryResultCacheWriter objects if subqueries are cached.

/// We should call finalize() on all of them.

for (auto & processor : *processors)

if (auto * stream_processor = dynamic_cast<StreamInQueryResultCacheTransform *>(&*processor); stream_processor)

stream_processor->finalizeWriteInQueryResultCache();

std::unique_ptr<SourceFromChunks> source,

std::unique_ptr<SourceFromChunks> source_totals,

std::unique_ptr<SourceFromChunks> source_extremes)

{

/// Construct the pipeline from the input source processors. The processors are provided by the query result cache to produce chunks of

/// a previous query result.

auto add_stream_from_query_result_cache_source = [&](OutputPort *& out_port, std::unique_ptr<SourceFromChunks> source_)

{

if (!source_)

return;

out_port = &source_->getPort();

processors->emplace_back(std::shared_ptr<SourceFromChunks>(std::move(source_)));

};

add_stream_from_query_result_cache_source(output, std::move(source));

add_stream_from_query_result_cache_source(totals, std::move(source_totals));

add_stream_from_query_result_cache_source(extremes, std::move(source_extremes));

{

if (!other.completed())

throw Exception(ErrorCodes::LOGICAL_ERROR, "Cannot add not completed pipeline");

resources.append(other.resources);

processors->insert(processors->end(), std::make_move_iterator(other.processors->begin()), std::make_move_iterator(other.processors->end()));

{

if (!other.completed())

throw Exception(ErrorCodes::LOGICAL_ERROR, "Cannot add not completed pipeline");

resources.append(other.resources);

processors->insert(processors->end(), other.processors->begin(), other.processors->end());

{

QueryPipeline to_remove = std::move(*this);

*this = QueryPipeline();

{

if (processors)

{

for (auto & processor : *processors)

processor->cancel();

}

{

if (port)

{

auto transform = std::make_shared<ExpressionTransform>(port->getSharedHeader(), actions);

connect(*port, transform->getInputPort());

port = &transform->getOutputPort();

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

}

{

if (!pulling())

throw Exception(ErrorCodes::LOGICAL_ERROR, "Pipeline must be pulling to convert header");

const auto & source_header = output->getHeader();

/// Prefer matching the source columns to the target structure by name, not by position.

/// This is used to read external dictionaries from a local ClickHouse source: the dictionary expects its

/// columns in keys-first order, but the source query may return them in a different order. Matching by name

/// reorders the columns correctly and keeps the local source consistent with the remote one, which already

/// matches by name (see `adaptBlockStructure` in `RemoteQueryExecutor`).

///

/// Matching by name is only possible when every target column is present in the source by name. When the

/// source query does not name its columns to match the target (e.g. `SELECT 1, 1`), keep the historical

/// positional matching of a local dictionary source, so that such dictionaries continue to load.

auto match_columns_mode = ActionsDAG::MatchColumnsMode::Name;

for (const auto & column : columns)

{

if (!source_header.has(column.name))

{

match_columns_mode = ActionsDAG::MatchColumnsMode::Position;

break;

}

}

auto converting = ActionsDAG::makeConvertingActions(

source_header.getColumnsWithTypeAndName(),

columns,

match_columns_mode,

context);

auto actions = std::make_shared<ExpressionActions>(std::move(converting));

addExpression(output, actions, *processors);

addExpression(totals, actions, *processors);

addExpression(extremes, actions, *processors);

{

auto callback = std::make_unique<ReadProgressCallback>();

callback->setProgressCallback(progress_callback);

callback->setQuota(quota);

callback->setNormalizedQueryHash(normalized_query_hash);

callback->setProcessListElement(process_list_element);

if (!update_profile_events)

callback->disableProfileEventUpdate();

return callback;