| 549 | } |
| 550 | |
| 551 | void ReadFromRemote::addLazyPipe( |
| 552 | Pipes & pipes, const ClusterProxy::SelectStreamFactory::Shard & shard, const SharedHeader & out_header, size_t parallel_marshalling_threads) |
| 553 | { |
| 554 | bool add_agg_info = stage == QueryProcessingStage::WithMergeableState; |
| 555 | bool add_totals = false; |
| 556 | bool add_extremes = false; |
| 557 | bool async_read = context->getSettingsRef()[Setting::async_socket_for_remote]; |
| 558 | const bool async_query_sending = context->getSettingsRef()[Setting::async_query_sending_for_remote]; |
| 559 | |
| 560 | if (stage == QueryProcessingStage::Complete) |
| 561 | { |
| 562 | if (const auto * ast_select = shard.query->as<ASTSelectQuery>()) |
| 563 | add_totals = ast_select->group_by_with_totals; |
| 564 | add_extremes = context->getSettingsRef()[Setting::extremes]; |
| 565 | } |
| 566 | |
| 567 | std::shared_ptr<const ActionsDAG> pushed_down_filters = filter_actions_dag; |
| 568 | |
| 569 | /// Override cluster_for_parallel_replicas to match the distributed table's cluster, |
| 570 | /// same as addPipe() does. Without this, the _shard_num scalar from the distributed |
| 571 | /// execution can mismatch the parallel replicas cluster shard count, causing a crash |
| 572 | /// in prepareClusterForParallelReplicas (STID 5066). |
| 573 | if (context->canUseTaskBasedParallelReplicas()) |
| 574 | { |
| 575 | if (context->getSettingsRef()[Setting::cluster_for_parallel_replicas].changed) |
| 576 | { |
| 577 | const String cluster_for_parallel_replicas = context->getSettingsRef()[Setting::cluster_for_parallel_replicas]; |
| 578 | if (cluster_for_parallel_replicas != cluster_name) |
| 579 | LOG_INFO( |
| 580 | log, |
| 581 | "cluster_for_parallel_replicas has been set for the query but has no effect: {}. Distributed table cluster is " |
| 582 | "used: {}", |
| 583 | cluster_for_parallel_replicas, |
| 584 | cluster_name); |
| 585 | } |
| 586 | |
| 587 | LOG_TRACE(log, "Setting `cluster_for_parallel_replicas` to {}", cluster_name); |
| 588 | context->setSetting("cluster_for_parallel_replicas", cluster_name); |
| 589 | } |
| 590 | |
| 591 | /// The storage is only consumed by the stale-replica branch below, which applies solely to |
| 592 | /// replicated tables. Table functions have an empty main table and reach this path only when the |
| 593 | /// use_delayed_remote_source failpoint forces a lazy read, and that branch is skipped for them, so |
| 594 | /// resolving the empty StorageID would needlessly throw. This mirrors the guard in addPipe. |
| 595 | StoragePtr storage; |
| 596 | if (!table_func_ptr) |
| 597 | { |
| 598 | const StorageID resolved_id = context->resolveStorageID(shard.main_table ? shard.main_table : main_table); |
| 599 | storage = DatabaseCatalog::instance().tryGetTable(resolved_id, context); |
| 600 | if (!storage) |
| 601 | throw Exception(ErrorCodes::UNKNOWN_TABLE, "Storage with id {} not found", resolved_id); |
| 602 | } |
| 603 | |
| 604 | auto lazily_create_stream = [ |
| 605 | my_shard = shard, my_shard_count = shard_count, my_distributed_fanout = shards.size(), |
| 606 | query = shard.query, header = shard.header, |
| 607 | my_context = context, my_throttler = throttler, |
| 608 | my_main_table = main_table, my_table_func_ptr = table_func_ptr, |
nothing calls this directly
no test coverage detected