The only way to use this feature is to specify an empty database name in Distributed table definition, but then you get a warning:

```
CREATE TABLE `demo_loan_all` (...) ENGINE=Distributed(test_cluster_two_shards_different_databases, '', demo_loan_01568, cityHash64(id));

<Warning> StorageDistributed (demo_loan_all): Name of remote database is empty. Default database will be used implicitly.
```
{
        if (settings.distributed_group_by_no_merge == DISTRIBUTED_GROUP_BY_NO_MERGE_AFTER_AGGREGATION)
            return QueryProcessingStage::WithMergeableStateAfterAggregation;
        else
            return QueryProcessingStage::Complete;
    }
    
    
    if (to_stage == QueryProcessingStage::WithMergeableState)
        return QueryProcessingStage::WithMergeableState;
    
    
    if (getClusterQueriedNodes(settings, cluster) == 1)
        return QueryProcessingStage::Complete;
    if (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))
    {
        Block sharding_key_block = sharding_key_expr->getSampleBlock();
        auto stage = getOptimizedQueryProcessingStage(query_info.query, settings.extremes, sharding_key_block);
        if (stage)
        {
            LOG_DEBUG(log, "Force processing stage to {}", QueryProcessingStage::toString(*stage));
            return *stage;
        }
    }
    return QueryProcessingStage::WithMergeableState;
}
Pipe StorageDistributed::read(