**How to reproduce**
Clickhouse version 21.2-21.3

```
SELECT x
FROM
(
    SELECT 1 AS x
    UNION DISTINCT
    SELECT 1
)
LIMIT 1

Query id: b82bbb8d-dea9-4f1c-9a2f-5b0e9b6c2422


0 rows in set. Elapsed: 0.002 sec.

Received exception from server (version 21.2.3):
Code: 49. DB::Exception: Received from localhost:9000. DB::Exception: Bad cast from type DB::ASTSelectWithUnionQuery to DB::ASTSelectQuery.
```


executeFetchColumns(from_stage, query_plan, expressions.prewhere_info, expressions.columns_to_remove_after_prewhere);
        LOG_TRACE(log, "{} -> {}", QueryProcessingStage::toString(from_stage), QueryProcessingStage::toString(options.to_stage));
    }
    if (options.to_stage > QueryProcessingStage::FetchColumns)
    {
        
        bool aggregate_overflow_row =
            expressions.need_aggregate &&
            query.group_by_with_totals &&
            settings.max_rows_to_group_by &&
            settings.group_by_overflow_mode == OverflowMode::ANY &&
            settings.totals_mode != TotalsMode::AFTER_HAVING_EXCLUSIVE;
        
        bool aggregate_final =
            expressions.need_aggregate &&
            options.to_stage > QueryProcessingStage::WithMergeableState &&
            !query.group_by_with_totals && !query.group_by_with_rollup && !query.group_by_with_cube;
        auto preliminary_sort = [&]()
        {
            /** For distributed query processing,
              *  if no GROUP, HAVING set,
              *  but there is an ORDER or LIMIT,
              *  then we will perform the preliminary sorting and LIMIT on the remote server.
              */
            if (!expressions.second_stage && !expressions.need_aggregate && !expressions.hasHaving())
            {
                if (expressions.has_order_by)
                    executeOrder(query_plan, query_info.input_order_info);
                if (expressions.has_order_by && query.limitLength())
                    executeDistinct(query_plan, false, expressions.selected_columns, true);