当碎片的中间结果与最终结果相结合时，问题似乎就出现了。
让我们检查每个shard的结果（将使用distributed\u group\u by\u no\u merge-setting来禁用合并每个shard的中间结果）：

select any(_shard_num), topK(30)(Country) 
from distributed_table
SETTINGS distributed_group_by_no_merge = 1

在每个shard上，topk函数都能正常工作，因此您可以手动组合所有中间结果：

SELECT arrayDistinct(
  arrayMap(x -> x.1, 
    /* sort values by frequency */
    arraySort(x -> x.2, 
      /* converts an array of arrays to a flat array */
      flatten(
        /* group results from shards to one array */
        groupArray(
          /* assign each value the index number */
          arrayMap((x, index) -> (x, index), shard_result, arrayEnumerate(shard_result))))))) ordered_value
FROM (
  select topK(30)(Country) AS shard_result
  from distributed_table
  SETTINGS distributed_group_by_no_merge = 1)