**Use case**

Run view on each shard individually and insert into local node.

**Describe the solution you'd like**

```sql

INSERT INTO FUNCTION cluster('all-sharded', map, output, key_2) SELECT *
FROM cluster('all-sharded', view(
    SELECT *
    FROM map.input
    WHERE (key_1 % 2) = toUInt32(getMacro('all-sharded-shard'))
))
SETTINGS parallel_distributed_insert_select = 2

Query id: 0ef3707f-2b74-4367-bf6d-82939a64e0f2


0 rows in set. Elapsed: 0.002 sec.

Received exception from server (version 21.11.10):
Code: 60. DB::Exception: Received from localhost:9000. DB::Exception: Both table name and UUID are empty. (UNKNOWN_TABLE)
```
{
    bool optimize_sharding_key_aggregation =
        settings.optimize_skip_unused_shards &&
        settings.optimize_distributed_group_by_sharding_key &&
        has_sharding_key &&
        (settings.allow_nondeterministic_optimize_skip_unused_shards || sharding_key_is_deterministic);
    QueryProcessingStage::Enum default_stage = QueryProcessingStage::WithMergeableStateAfterAggregation;
    if (settings.distributed_push_down_limit)
        default_stage = QueryProcessingStage::WithMergeableStateAfterAggregationAndLimit;
    const auto & select = query_info.query->as<ASTSelectQuery &>();
    auto expr_contains_sharding_key = [&](const auto & exprs) -> bool
    {
        std::unordered_set<std::string> expr_columns;
        for (auto & expr : exprs)
        {
            auto id = expr->template as<ASTIdentifier>();
            if (!id)
                continue;
            expr_columns.emplace(id->name());
        }
        for (const auto & column : sharding_key_expr->getRequiredColumns())
        {
            if (!expr_columns.contains(column))