20.10.1.4853

```sql
drop table if exists xp;
drop table if exists xp_d;

create table xp(A Date, B Int64, S String) Engine=MergeTree partition by toYYYYMM(A) order by B;
insert into xp select '2020-01-01', number , '' from numbers(100000);
insert into xp select '2020-10-01', number , '' from numbers(100000);
create table xp_d as xp Engine=Distributed(test_shard_localhost, currentDatabase(), xp);

select count() from xp_d prewhere toYYYYMM(A) global in (select toYYYYMM(min(A)) from xp_d) where B > -1;
┌─count()─┐
│       0 │     <------ expected 100000
└─────────┘

-- global in & both predicates in prewhere 
select count() from xp_d prewhere toYYYYMM(A) global in (select toYYYYMM(min(A)) from xp_d) and B > -1;
┌─count()─┐
│  100000 │
└─────────┘

-- global in & both predicates in where 
select count() from xp_d where toYYYYMM(A) global in (select toYYYYMM(min(A)) from xp_d) and B > -1;
┌─count()─┐
│  100000 │
└─────────┘

-- not global
select count() from xp_d prewhere toYYYYMM(A) in (select toYYYYMM(min(A)) from xp_d) where B > -1;
┌─count()─┐
│  100000 │
└─────────┘



select count() from xp_d prewhere toYYYYMM(A) global in (select 202001) where B > -1;
┌─count()─┐
│       0 │
└─────────┘
```

hmm, it worked in 19.13.7.57
if (group_by_info && settings.optimize_aggregation_in_order)
        group_by_sort_description = getSortDescriptionFromGroupBy(getSelectQuery());
    else
        group_by_info = nullptr;
    auto merge_threads = max_streams;
    auto temporary_data_merge_threads = settings.aggregation_memory_efficient_merge_threads
                                        ? static_cast<size_t>(settings.aggregation_memory_efficient_merge_threads)
                                        : static_cast<size_t>(settings.max_threads);
    bool storage_has_evenly_distributed_read = storage && storage->hasEvenlyDistributedRead();
    auto aggregating_step = std::make_unique<AggregatingStep>(
            query_plan.getCurrentDataStream(),
            params, final,
            settings.max_block_size,
            merge_threads,
            temporary_data_merge_threads,
            storage_has_evenly_distributed_read,
            std::move(group_by_info),
            std::move(group_by_sort_description));
    query_plan.addStep(std::move(aggregating_step));
}
void InterpreterSelectQuery::executeMergeAggregated(QueryPlan & query_plan, bool overflow_row, bool final)
{
    const auto & header_before_merge = query_plan.getCurrentDataStream().header;
    ColumnNumbers keys;
    for (const auto & key : query_analyzer->aggregationKeys())
        keys.push_back(header_before_merge.getPositionByName(key.name));
    /** There are two modes of distributed aggregation.
      *
      * 1. In different threads read from the remote servers blocks.
      * Save all the blocks in the RAM. Merge blocks.