Construindo uma busca por filtros: quando uma máquina não basta

Distribuir não torna uma operação cara mais barata por definição. Ele divide estado e trabalho entre máquinas, mas acrescenta fan-out, rede, falhas parciais e coordenação de versões.

Por isso, nesta conversa a distribuição chega tarde. Antes dela, já sabemos o que cada partição precisa fazer: filtrar, contar facetas, ordenar, paginar e publicar snapshots. A ideia de particionar horizontalmente e executar operadores perto dos dados tem uma linhagem explícita em arquiteturas paralelas shared-nothing.[2]

Scatter distribui a pergunta; gather combina respostas que pertencem ao mesmo contrato.

Particionar itens, não consultas

Cada item pertence a uma única partição primária. Podemos calcular ou consultar essa localização a partir de um mapa versionado:

placement(item_id, epoch) → shard

Dentro do shard, os itens recebem IDs locais densos. Assim, cada bitmap cobre apenas a população daquela partição.

Particionar por item simplifica updates e contagens globais:

• um update possui um owner;
• shards são disjuntos;
• contagens podem ser somadas;
• resultados locais podem ser combinados.

O custo é que uma query arbitrária não sabe em qual shard estão seus resultados. No caso geral, ela precisa consultar todos.

Se país, região ou família de categoria aparecem em quase toda consulta, podemos usá-los como domínio semântico de primeiro nível e distribuir os itens por hash dentro desse domínio:

routing_key = (país, família_de_categoria, hash(product_group_id))

A dimensão semântica poda shards. O hash evita concentrar uma marca ou vendedor. As variações do mesmo produto permanecem juntas para permitir agrupamento local. Essa escolha não é gratuita: categorias quentes, itens que mudam de domínio e consultas sem a chave semântica precisam de tratamento explícito.

Eu criaria muito mais partições lógicas do que servidores físicos. Um servidor hospeda várias unidades; o mapa versionado decide onde cada uma vive. Isso permite mover pequenas parcelas sem reorganizar o índice inteiro.

Micro-partições preservam a identidade

Um shard lógico pode ser apenas um agrupamento de micro-partições. O ID de busca passa a ser:

SearchDocumentID = (micro_partition_id, local_slot)

O split troca ownership no mapa; os postings das unidades pequenas continuam válidos.

Ao dividir o shard 17, as micro-partições 100–124 passam ao 17A e 125–149, ao 17B. Na maioria dos casos movemos arquivos e metadados, não renumeramos cada documento. Requisições destinadas a várias micro-partições na mesma máquina continuam agrupadas numa única mensagem.

O caminho da query distribuída

O coordenador executa cinco tarefas. A separação entre um data plane que encaminha/executa e um control plane que observa carga e decide placement fora do caminho crítico é a mesma distinção central do Slicer.[3]

1. normalizar: ordenar valores, converter unidades e produzir uma chave canônica;
2. rotear: consultar o mapa e podar domínios impossíveis;
3. disparar: agrupar subconsultas por máquina com request ID, epoch e deadline;
4. executar localmente: filtrar, contar e produzir top k sem enviar bitmaps;
5. combinar: somar contagens e fazer merge das sequências locais.

A regra é mover a consulta até os índices, não transportar os maiores conjuntos intermediários pela rede.

A multiplicação escondida

Com Q queries externas por segundo e S shards:

shard_queries_per_second = Q × S

Dez mil queries por segundo e 64 shards produzem 640 mil shard-queries por segundo antes de réplicas, retries ou consultas especulativas.

Facet counts também podem dominar a rede. Se cada um dos 64 shards devolver 300 contadores de 64 bits, são pelo menos 153.600 bytes por query externa apenas em contadores. A dez mil queries por segundo, isso representa 1,536 GB/s — 12,288 Gbit/s — antes de IDs, top k, headers, serialização e réplicas. O protocolo precisa limitar quais facetas são pedidas e evitar devolver vetores densos quando a interface não os usará.

Essa conta muda a topologia. O coordenador precisa:

• reutilizar conexões;
• impor deadline fim a fim;
• limitar concorrência por origem e classe de query;
• cancelar trabalho que perdeu valor;
• evitar retries independentes em várias camadas;
• observar goodput, não apenas requests recebidos.

Adicionar shards reduz itens por máquina e aumenta fan-out. Existe uma fronteira, não uma direção única chamada “mais horizontal”.

O contrato de uma shard-query

O coordenador envia:

query normalizada
ordenação
k
facetas solicitadas
snapshot ou epoch mínima
budget de tempo e bytes
request_id

Cada shard devolve:

geração lida
top k local
contagem total local
contagens locais por faceta
estado: complete | timeout | unavailable
custo observado

O request_id permite correlacionar e descartar respostas tardias. A deadline original atravessa todas as chamadas; um shard não deveria continuar consumindo CPU para uma resposta que o coordenador já abandonou.

Somar contagens é correto sob condições

Se cada item pertence exatamente a um shard e todos respondem sobre um snapshot compatível:

count_global = Σ count_shard

A soma deixa de ser confiável quando:

• um shard está ausente;
• dois shards acreditam possuir o mesmo intervalo durante rebalanceamento;
• uma partição respondeu na geração anterior e outra na nova;
• uma resposta parcial é apresentada como completa.

O protocolo precisa escolher. Pode falhar a query, retornar estado parcial explicitamente ou usar um snapshot global coordenado. O que não pode fazer é omitir a incompletude.

Um único número generation=42 em todo o cluster é simples de entender, mas pode atrasar publicação pelo shard mais lento. Um vetor de gerações permite progresso independente, mas aumenta o estado do cursor e a complexidade de reprodução. Novamente, consistência tem custo material.

Top k global a partir de top k local

Para uma mesma ordem total, é suficiente pedir os primeiros k itens de cada shard. Qualquer item fora do top k local possui pelo menos k itens melhores no próprio shard e, portanto, não pode entrar no top k global.

O coordenador faz um merge de S sequências ordenadas, normalmente com um heap de tamanho S:

custo de merge ≈ O(k log S)

O argumento depende de todos usarem a mesma chave, desempate e snapshot. Se a ordenação contém uma função contextual calculada de modo diferente por shard, a propriedade se perde.

Para paginação, o cursor precisa carregar posição global e identidade do snapshot. O coordenador envia a âncora para todos os shards; cada um retorna seus próximos candidatos locais; o merge produz a página seguinte.

Réplicas não são apenas cópias

Cada shard precisa de réplicas para sobreviver à perda de uma máquina e dividir leituras. O mapa de placement indica primário, réplicas e epoch.

Uma réplica só pode atender uma query se souber:

• até qual sequência de updates aplicou;
• quais segmentos e manifesto formam sua geração;
• se ainda possui autorização para servir aquele intervalo;
• quanto atraso existe em relação ao primário.

Enviar a leitura para a réplica mais rápida sem freshness mínima pode reduzir latência e devolver item vendido. Enviar somente ao primário simplifica consistência e concentra carga. A política pode variar por tipo de filtro ou risco, desde que apareça no contrato.

Três operações resolvem gargalos diferentes. Tratar shard, réplica, placement e evento planejado como objetos de um framework próprio é também a motivação operacional do Shard Manager.[4]

Split aumenta paralelismo local, mas também aumenta fan-out. Adicionar réplica divide leitura, mas não reduz o trabalho de uma query. Mover shard corrige placement físico sem alterar a partição lógica.

Rebalanceamento sem dupla autoridade

Ao mover um intervalo de A para B:

1. B copia um snapshot de A
2. B aplica o log posterior ao snapshot
3. o controlador verifica que B alcançou o watermark
4. publica um novo mapa com epoch maior
5. routers passam a enviar novas operações para B
6. A permanece como fallback por uma janela delimitada
7. A remove o intervalo depois da convergência

Requests carregam a epoch conhecida. Um owner antigo responde com o mapa mais novo ou redireciona; não aceita silenciosamente uma escrita que criaria duas autoridades.

Hashing consistente é uma família possível para limitar movimento quando o conjunto de máquinas muda.[1] Um mapa explícito também pode funcionar. O mecanismo escolhido precisa representar pesos, failure domains, versionamento e rollback; “hash” sozinho não fecha placement operacional.

Falhas parciais e cauda

Uma query distribuída termina na velocidade de sua resposta mais lenta, a menos que o protocolo faça outra escolha. Em fan-outs amplos, a cauda de componentes individuais passa a dominar a experiência agregada.[5] Precisamos medir:

• latência por shard e por fase;
• fan-out efetivo;
• bytes enviados e recebidos;
• fila e CPU por tipo de query;
• shards ausentes ou stale;
• respostas que chegaram depois da deadline;
• diferença entre resultados completos e parciais.

Hedging pode reduzir cauda enviando a mesma leitura a outra réplica, mas multiplica carga justamente durante degradação. Retry sem budget pode transformar um shard lento em uma cascata. O controle pertence ao coordenador, que conhece a deadline fim a fim.

Ao final desta parte, a busca funciona em várias máquinas. Ainda falta explicar como todas recebem updates, recuperam estado depois de crash, controlam sobrecarga e provam que o índice distribuído continua equivalente ao modelo simples.

Referências

1. David Karger et al. Consistent Hashing and Random Trees, STOC 1997.
2. David DeWitt et al. The Gamma Database Machine Project, IEEE 1990.
3. Atul Adya et al. Slicer: Auto-Sharding for Datacenter Applications, OSDI 2016.
4. Chunqiang Tang et al. Shard Manager: A Generic Shard Management Framework for Geo-distributed Applications, SOSP 2021.
5. Jeffrey Dean e Luiz André Barroso. The Tail at Scale, Communications of the ACM, 2013.

Parte anterior: Ordenar e paginar

Próxima parte: Fechar o contrato online