PolarDB:Implementação do operador TopK no índice columnstore

ORDER BY column LIMIT offset, count

Fila de prioridade (TopK baseado em heap)

O sistema mantém um max-heap de tamanho K. Para cada registro verificado, ele confere se o registro pertence aos K primeiros; caso positivo, substitui o elemento do topo e reequilibra o heap. Após uma varredura completa, o heap contém os K maiores registros.

Essa abordagem funciona bem quando K é pequeno. Com valores grandes de K — como 1.000.100 — surgem dois problemas:

• Pressão de memória: o heap pode não caber na memória.

• Ineficiência de cache: operações de heap exigem acesso aleatório à memória. Quando K é grande, isso causa cache misses frequentes e reduz o throughput.

Truncamento durante merge sort

PolarDB, ClickHouse, SQL Server e DuckDB utilizam esta abordagem. O sistema gera runs ordenadas durante a classificação e trunca cada run mesclada para offset + limit registros. Apenas os registros em [offset, offset + limit) importam, eliminando a necessidade de ordenar todos os dados.

Esse método escala para disco quando a memória é insuficiente. No entanto, o truncamento só ajuda quando uma run ordenada atinge comprimento suficiente. Em deep paging com K grande, as runs ordenadas no início da mesclagem costumam ser menores que offset + limit, obrigando a ordenação completa do conjunto de dados antes que o truncamento surta efeito.

Filtro de entrada autoajustável

Goetz Graefe propôs inicialmente esta técnica, posteriormente adotada pelo ApsaraDB for ClickHouse. O sistema mantém um valor de corte — um limite superior para os registros que podem aparecer no resultado TopK. Registros acima do corte são excluídos antes de entrarem em uma run ordenada. Após a construção de cada run, se o comprimento for maior que K, o K-ésimo registro torna-se o novo corte. Como o novo corte é sempre menor ou igual ao anterior, o filtro se ajusta continuamente.

Exemplo (K = 3):

Diferentemente das operações de heap, o filtro autoajustável acessa a memória sequencialmente tanto na filtragem por corte quanto na acumulação de runs ordenadas. Isso evita a penalidade de acesso aleatório inerente à manutenção de heaps.

Algoritmo em memória: filtro de entrada autoajustável acelerado por SIMD

Quando há memória suficiente, o IMCI usa o filtro de entrada autoajustável em vez de uma fila de prioridade. Motivos para evitar filas de prioridade com K grande:

• O acesso aleatório à memória durante a manutenção do heap degrada o desempenho quando K é grande.

• O tamanho do heap deve ser igual a K, fazendo a pressão de memória crescer linearmente com K.

O filtro autoajustável evita ambos os problemas:

• Tanto a filtragem por corte quanto a acumulação de runs ordenadas acessam a memória sequencialmente.

• O filtro funciona corretamente para qualquer valor de K, cobrindo paginação rasa e profunda sem condição de contorno.

Aceleração SIMD: a filtragem por corte é simples, repetitiva e frequente — comparando cada registro com o corte atual. O IMCI acelera esse processo usando instruções SIMD (single instruction multiple data), que aplicam a mesma comparação a vários registros em paralelo. O filtro reutiliza a infraestrutura de avaliação de expressões do predicado de varredura de tabela, eliminando a necessidade de um caminho de código adicional.

Algoritmo em disco: poda baseada em ZoneMap com merge sort

Quando a memória é insuficiente, o IMCI usa merge sort com truncamento. Razões para não usar o filtro autoajustável em disco:

• Salvar runs ordenadas acumuladas no disco e executar ordenação externa durante a pré-mesclagem gera grande volume de I/O de disco.

• Para K grande, a pré-mesclagem pode processar uma quantidade significativa de dados fora de [offset, offset + limit) antes que um corte útil seja estabelecido.

O merge sort com truncamento evita esses problemas, e a poda baseada em ZoneMap elimina ainda mais I/O ao usar estatísticas de mínimo/máximo das runs ordenadas para pular runs que não contribuem para o resultado.

Funcionamento da poda por ZoneMap:

Cada run ordenada armazena seus valores mínimo e máximo. Um valor de barreira divide todas as runs ordenadas em três tipos:

Duas regras de poda eliminam runs ordenadas que não afetam o resultado:

• Se o total de registros nos Tipos A + B < offset, todos os registros do Tipo A caem em [0, offset). Exclua o Tipo A das mesclagens subsequentes.

• Se o total de registros no Tipo A > offset + limit, todos os registros do Tipo C caem em [offset + limit, N). Exclua o Tipo C das mesclagens subsequentes.

Processo de poda:

1. Construa um ZoneMap contendo os valores mínimo e máximo de cada run ordenada.

2. Encontre a Barreira 1 (a maior possível) onde os registros nos Tipos A + B < offset.

3. Encontre a Barreira 2 (a menor possível) onde os registros no Tipo A > offset + limit.

4. Use a Barreira 1 e a Barreira 2 para excluir as runs ordenadas correspondentes das mesclagens subsequentes.

Seleção dinâmica de algoritmo

Em vez de usar um limiar fixo de K, o IMCI seleciona o algoritmo dinamicamente por meio de um mecanismo de fallback:

1. Sempre inicie com o algoritmo em memória.

2. Se a memória permanecer suficiente durante todo o processo, conclua o cálculo em memória.

3. Se a memória acabar — por exemplo, espaço insuficiente para armazenar em cache runs ordenadas suficientes contendo mais de K registros, ou espaço insuficiente para concluir a pré-mesclagem — acione o fallback:

   • Colete os valores min/max das runs ordenadas em memória e construa um ZoneMap.

   • Salve essas runs ordenadas em disco.

   • Mude para o algoritmo em disco para o restante do cálculo.

4. Conclua o cálculo usando o algoritmo em disco.

Ambos os algoritmos usam as mesmas estruturas de dados, portanto o fallback não exige reorganização de dados. As runs ordenadas acumuladas durante a fase em memória são usadas diretamente pelo algoritmo em disco, sem perda de precisão.

Materialização tardia

Ao construir runs ordenadas, o IMCI materializa apenas IDs de linha e as colunas ou expressões referenciadas em ORDER BY. As colunas de saída são buscadas no armazenamento após a definição do conjunto de resultados TopK.

Isso reduz a sobrecarga de duas formas:

• IDs de linha são compactos, permitindo que mais registros caibam no mesmo orçamento de memória e estendendo a faixa de aplicação do algoritmo em memória.

• Os registros são reordenados frequentemente durante o cálculo TopK por meio de operações de cópia e troca. Materializar apenas IDs de linha minimiza o custo por registro dessas operações.

O compromisso: buscar colunas de saída por ID de linha após a ordenação pode exigir I/O aleatório. Em cenários de deep paging, o conjunto de resultados real é pequeno (por exemplo, 100 registros), tornando esse I/O aleatório desprezível.

Pushdown de computação

Durante a execução do filtro autoajustável, o valor de corte atual é enviado ao operador de varredura de tabela como um novo predicado. A varredura de tabela pode então aplicar esse predicado usando a infraestrutura de pruner existente, filtrando no nível de pacote ou grupo de linhas antes que os dados cheguem ao operador TopK.

Essa técnica reduz a sobrecarga de duas maneiras:

• Redução de I/O: pacotes ou grupos de linhas que contêm apenas registros acima do corte são totalmente ignorados.

• Redução de computação: pacotes ou grupos de linhas filtrados não são processados pelos operadores de camadas superiores.

SELECT
    l_orderkey,
    sum(l_quantity)
FROM
    lineitem
GROUP BY
    l_orderkey
ORDER BY
    sum(l_quantity) DESC
LIMIT
    1000000, 100;