Backport #16993 to 20.12: Fix Merge(Distributed()) with JOIN

src/Storages/StorageMerge.cpp

@@ -135,6 +135,18 @@ bool StorageMerge::mayBenefitFromIndexForIn(const ASTPtr & left_in_operand, cons
 
 QueryProcessingStage::Enum StorageMerge::getQueryProcessingStage(const Context & context, QueryProcessingStage::Enum to_stage, SelectQueryInfo & query_info) const
 {
+    ASTPtr modified_query = query_info.query->clone();
+    auto & modified_select = modified_query->as<ASTSelectQuery &>();
+    /// In case of JOIN the first stage (which includes JOIN)
+    /// should be done on the initiator always.
+    ///
+    /// Since in case of JOIN query on shards will receive query w/o JOIN (and their columns).
+    /// (see modifySelect()/removeJoin())
+    ///
+    /// And for this we need to return FetchColumns.
+    if (removeJoin(modified_select))
+        return QueryProcessingStage::FetchColumns;
+
     auto stage_in_source_tables = QueryProcessingStage::FetchColumns;
 
     DatabaseTablesIteratorPtr iterator = getDatabaseIterator(context);

tests/queries/0_stateless/01560_merge_distributed_join.sql

+-- test from https://github.com/ClickHouse/ClickHouse/issues/11755#issuecomment-700850254
+DROP TABLE IF EXISTS cat_hist;
+DROP TABLE IF EXISTS prod_hist;
+DROP TABLE IF EXISTS products_l;
+DROP TABLE IF EXISTS products;
+
+CREATE TABLE cat_hist (categoryId UUID, categoryName String) ENGINE Memory;
+CREATE TABLE prod_hist (categoryId UUID, productId UUID) ENGINE = MergeTree ORDER BY productId;
+
+CREATE TABLE products_l AS prod_hist ENGINE = Distributed(test_cluster_two_shards, currentDatabase(), prod_hist);
+CREATE TABLE products as prod_hist ENGINE = Merge(currentDatabase(), '^products_');
+
+SELECT * FROM products AS p LEFT JOIN cat_hist AS c USING (categoryId);
+SELECT * FROM products AS p GLOBAL LEFT JOIN cat_hist AS c USING (categoryId);
+
+DROP TABLE cat_hist;
+DROP TABLE prod_hist;
+DROP TABLE products_l;
+DROP TABLE products;