{

const auto & storage = table_node.getStorage();

if (!storage->isMergeTree() && !typeid_cast<const StorageDummy *>(storage.get()))

return false;

if (!storage->supportsReplication() && !context->getSettingsRef()[Setting::parallel_replicas_for_non_replicated_merge_tree])

return false;

/// Parallel replicas not supported with FINAL.

if (table_node.hasTableExpressionModifiers() && table_node.getTableExpressionModifiers()->hasFinal())

return false;

return true;

{

std::vector<const QueryNode *> res;

while (query_tree_node)

{

auto join_tree_node_type = query_tree_node->getNodeType();

switch (join_tree_node_type)

{

case QueryTreeNodeType::TABLE:

{

const auto & table_node = query_tree_node->as<TableNode &>();

if (canUseTableForParallelReplicas(table_node, context))

return res;

return {};

}

case QueryTreeNodeType::TABLE_FUNCTION:

{

return {};

}

case QueryTreeNodeType::QUERY:

{

const auto & query_node_to_process = query_tree_node->as<QueryNode &>();

query_tree_node = query_node_to_process.getJoinTree().get();

res.push_back(&query_node_to_process);

break;

}

case QueryTreeNodeType::UNION:

{

const auto & union_node = query_tree_node->as<UnionNode &>();

const auto & union_queries = union_node.getQueries().getNodes();

if (union_queries.empty())

return {};

query_tree_node = union_queries.front().get();

break;

}

case QueryTreeNodeType::ARRAY_JOIN:

{

const auto & array_join_node = query_tree_node->as<ArrayJoinNode &>();

query_tree_node = array_join_node.getTableExpression().get();

break;

}

case QueryTreeNodeType::CROSS_JOIN:

{

/// TODO: We can parallelize one table

return {};

}

case QueryTreeNodeType::JOIN:

{

const auto & join_node = query_tree_node->as<JoinNode &>();

const auto join_kind = join_node.getKind();

const auto join_strictness = join_node.getStrictness();

if (join_kind == JoinKind::Left || (join_kind == JoinKind::Inner && join_strictness == JoinStrictness::All))

query_tree_node = join_node.getLeftTableExpression().get();

else if (join_kind == JoinKind::Right && join_strictness != JoinStrictness::RightAny)

query_tree_node = join_node.getRightTableExpression().get();

else

return {};

break;

}

default:

{

throw Exception(ErrorCodes::LOGICAL_ERROR,

"Unexpected node type for table expression. "

"Expected table, table function, query, union, join or array join. Actual {}",

query_tree_node->getNodeTypeName());

}

}

}

return res;

{

using Base = InDepthQueryTreeVisitorWithContext<ReplaceTableNodeToDummyVisitor>;

using Base::Base;

void enterImpl(QueryTreeNodePtr & node)

{

auto * table_node = node->as<TableNode>();

auto * table_function_node = node->as<TableFunctionNode>();

if (table_node || table_function_node)

{

const auto & storage_snapshot = table_node ? table_node->getStorageSnapshot() : table_function_node->getStorageSnapshot();

const auto & storage = storage_snapshot->storage;

auto storage_dummy = std::make_shared<StorageDummy>(

storage.getStorageID(),

/// To preserve information about alias columns, column description must be extracted directly from storage metadata.

storage_snapshot->metadata->getColumns(),

storage_snapshot,

storage.supportsReplication());

auto dummy_table_node = std::make_shared<TableNode>(std::move(storage_dummy), getContext());

if (table_node && table_node->hasTableExpressionModifiers())

dummy_table_node->getTableExpressionModifiers() = table_node->getTableExpressionModifiers();

dummy_table_node->setAlias(node->getAlias());

replacement_map.emplace(node.get(), std::move(dummy_table_node));

}

}

std::unordered_map<const IQueryTreeNode *, QueryTreeNodePtr> replacement_map;

{

ReplaceTableNodeToDummyVisitor visitor(context);

visitor.visit(query);

return query->cloneAndReplace(visitor.replacement_map);

{

std::ranges::reverse_view rv{stack};

for (const auto * node : rv)

LOG_DEBUG(getLogger(__PRETTY_FUNCTION__), "{}\n{}", CityHash_v1_0_2::Hash128to64(node->getTreeHash()), node->dumpTree());

{

const QueryPlan::Node * node = nullptr;

/// Below we will check subqueries from `stack` to find outermost subquery that could be executed remotely.

/// Currently traversal algorithm considers only steps with 0 or 1 children and JOIN specifically.

/// When we found some step that requires finalization on the initiator (e.g. GROUP BY) there are two options:

/// 1. If plan looks like a single path (e.g. AggregatingStep -> ExpressionStep -> Reading) we can execute

/// current subquery as a whole with replicas.

/// 2. If we were inside JOIN we cannot offload the whole subquery to replicas because at least one side

/// of the JOIN needs to be finalized on the initiator.

/// So this flag is used to track what subquery to return once we hit a step that needs finalization.

bool inside_join = false;

const QueryNode & query_node,

const PlannerContextPtr & planner_context,

std::shared_ptr<const StorageLimitsList> storage_limits)

{

auto processed_stage = QueryProcessingStage::WithMergeableState;

auto context = planner_context->getQueryContext();

QueryTreeNodePtr modified_query_tree = query_node.clone();

auto initial_header = InterpreterSelectQueryAnalyzer::getSampleBlock(

modified_query_tree, context, SelectQueryOptions(processed_stage).analyze());

rewriteJoinToGlobalJoin(modified_query_tree, context);

modified_query_tree = buildQueryTreeForShard(planner_context, modified_query_tree, /*allow_global_join_for_right_table*/ true);

ASTPtr modified_query_ast = queryNodeToDistributedSelectQuery(modified_query_tree);

auto header = InterpreterSelectQueryAnalyzer::getSampleBlock(

modified_query_tree, context, SelectQueryOptions(processed_stage).analyze());

const TableNode * table_node = findTableForParallelReplicas(modified_query_tree.get(), context);

if (!table_node)

throw Exception(ErrorCodes::LOGICAL_ERROR, "Can't determine table for parallel replicas");

QueryPlan query_plan;

ClusterProxy::executeQueryWithParallelReplicas(

query_plan,

table_node->getStorageID(),

header,

processed_stage,

modified_query_ast,

context,

storage_limits,

nullptr);

auto converting = ActionsDAG::makeConvertingActions(

header->getColumnsWithTypeAndName(),

initial_header->getColumnsWithTypeAndName(),

ActionsDAG::MatchColumnsMode::Position);

/// initial_header is a header expected by initial query.

/// header is a header which is returned by the follower.

/// They are different because tables will have different aliases (e.g. _table1 or _table5).

/// Here we just rename columns by position, with the hope the types would match.

auto step = std::make_unique<ExpressionStep>(query_plan.getCurrentHeader(), std::move(converting));

step->setStepDescription("Convert distributed names");

query_plan.addStep(std::move(step));

return {std::move(query_plan), std::move(processed_stage), {}, {}, {}};